Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The policy seam: one trait, one built-in engine, every engine an addon
3 : //! (ADR-0020).
4 : //!
5 : //! The broker takes no authorization decision of its own. It asks the
6 : //! engine a deployment gave it and obeys, and what an engine may answer is
7 : //! deliberately narrow: allow, deny, or narrow the operation. Nothing here
8 : //! parses a credential or validates a token — authentication, rate limiting
9 : //! and quotas stay in the gateway in front of the broker.
10 : //!
11 : //! Everything in this module is core code, compiled and tested in every
12 : //! build. The only engine the broker ships is [`AllowAll`], and conformance
13 : //! is asserted against it; any other engine is an addon crate outside
14 : //! `crates/`, behind an off-by-default `antares-broker` feature.
15 : //!
16 : //! The seam fails closed: an engine that errors, panics or runs past
17 : //! [`TIMEOUT`] denies. A deployment that wires in a broken engine loses
18 : //! service, never its access rules.
19 :
20 : use antares_ql::geo::GeoQuery;
21 : use antares_ql::QNode;
22 : use axum::http::HeaderMap;
23 : use serde_json::Value;
24 : use std::borrow::Cow;
25 : use std::collections::BTreeMap;
26 : use std::future::Future;
27 : use std::pin::Pin;
28 : use std::sync::LazyLock;
29 :
30 : /// The `@context` [`Filter::project`] reads rule conditions with.
31 : static CORE_CONTEXT: LazyLock<antares_jsonld::Context> =
32 : LazyLock::new(antares_jsonld::core_context);
33 : use std::time::Duration;
34 :
35 : /// What [`PolicyEngine::decide`] hands back. Boxed because an engine that
36 : /// asks a policy server has to await, and the trait must stay object-safe:
37 : /// the broker holds one `Arc<dyn PolicyEngine>` chosen at startup.
38 : pub type DecisionFuture<'a> = Pin<Box<dyn Future<Output = Decision> + Send + 'a>>;
39 :
40 : /// What [`PolicyEngine::decide_each`] hands back: one [`Decision`] per
41 : /// Operation it was given, in the same order.
42 : pub type DecisionsFuture<'a> = Pin<Box<dyn Future<Output = Vec<Decision>> + Send + 'a>>;
43 :
44 : /// How long an engine has to answer before the broker stops waiting and
45 : /// denies. Deployment knob (`ANTARES_POLICY_TIMEOUT_MS`), read once at
46 : /// first use. The default is short on purpose: the seam sits in front of
47 : /// every request, and an engine that cannot answer inside it is an outage
48 : /// either way — failing closed at 250 ms is the difference between a 403
49 : /// and a broker that stops accepting.
50 38 : pub static TIMEOUT: LazyLock<Duration> = LazyLock::new(|| {
51 38 : let env = std::env::var("ANTARES_POLICY_TIMEOUT_MS").ok();
52 38 : Duration::from_millis(timeout_ms(env.as_deref()))
53 38 : });
54 :
55 : /// `ANTARES_POLICY_TIMEOUT_MS` in milliseconds: a positive integer, else 250.
56 58 : fn timeout_ms(env: Option<&str>) -> u64 {
57 58 : env.and_then(|v| v.parse().ok())
58 58 : .filter(|n: &u64| *n > 0)
59 58 : .unwrap_or(250)
60 58 : }
61 :
62 : /// The reasons the seam denies on its own, rather than on an engine's word.
63 : /// Fixed strings: they reach the client in a ProblemDetails `detail`, and
64 : /// an engine's own text about why it failed is not the client's business.
65 : pub const ENGINE_TIMED_OUT: &str = "the policy engine did not answer in time";
66 : pub const ENGINE_FAILED: &str = "the policy engine failed";
67 :
68 : /// The clauses whose operations act on everything the tenant holds. There
69 : /// is no narrowed form of "delete every entity" or "snapshot the tenant",
70 : /// so a [`Decision::Filter`] on one of them is answered as a deny
71 : /// ([`resolve`]).
72 : pub const WHOLE_TENANT: [&str; 4] = ["5.6.21", "5.16.1", "5.16.2", "5.16.7"];
73 :
74 : /// The clauses whose handlers read the [`Filter`] an engine returns: the
75 : /// reads that can serve less than they were asked for. Every other clause
76 : /// takes its operation whole — a create either writes the Entity or does
77 : /// not — so a [`Decision::Filter`] there would be dropped, and the broker
78 : /// would perform in full an operation the engine believed it had narrowed.
79 : /// That is the one direction the seam may not fail in, so those clauses
80 : /// answer a narrowing as a deny ([`resolve`]).
81 : /// Discovery (5.7.5 to 5.7.10) is among them: the narrowing rules compute it
82 : /// from the Entities the narrowing selects.
83 : /// Query Subscriptions (5.8.4, 5.11.5) and Query Registrations (5.10.2)
84 : /// narrow the list ([`RECORD_LISTS`]).
85 : pub const FILTERABLE: [&str; 15] = [
86 : "5.7.1", "5.7.2", "5.7.3", "5.7.4", "5.7.5", "5.7.6", "5.7.7", "5.7.8", "5.7.9", "5.7.10",
87 : "5.14.4", "5.14.5", "5.8.4", "5.11.5", "5.10.2",
88 : ];
89 :
90 : /// The FILTERABLE clauses that list Subscriptions or Registrations. A
91 : /// narrowing there selects which documents are listed, counted and paged,
92 : /// and projects nothing: those are the client's own 5.2.12 and 5.2.9
93 : /// records, not Entities with Attributes to drop, so a [`Filter`] that
94 : /// projects is a deny there ([`resolve`]).
95 : pub const RECORD_LISTS: [&str; 3] = ["5.8.4", "5.11.5", "5.10.2"];
96 :
97 : /// List and Serve @context (5.13.3, 5.13.4). A stored `@context` may define
98 : /// a term for an Attribute the caller may not see, so a narrowing there,
99 : /// whatever it holds, says the caller sees fewer names than the tenant
100 : /// holds: the list is the core `@context` alone and a stored one is not
101 : /// found. [`resolve`] passes it through.
102 : pub const NAME_READS: [&str; 2] = ["5.13.3", "5.13.4"];
103 :
104 : /// Who is asking. The headers are the ones a deployment names for the
105 : /// seam to carry, copied verbatim and never interpreted: the broker does
106 : /// not know what a token is. They never leave this process — stripped from every
107 : /// forwarded request, absent from notifications, dead letters and logs,
108 : /// which is why [`std::fmt::Debug`] here prints names and not values.
109 : #[derive(Clone)]
110 : pub struct Subject {
111 : pub tenant: antares_model::TenantId,
112 : pub headers: Vec<(String, String)>,
113 : }
114 :
115 : /// The name of the reserved pair in [`Subject::headers`] that says which view
116 : /// the request came through (ADR-0020 amendment). `@` is no
117 : /// header-name character (RFC 9110 `tchar`), so no request can carry it:
118 : /// only [`through_view`] puts it there.
119 : pub const VIEW_PAIR: &str = "@view";
120 :
121 : tokio::task_local! {
122 : /// The view the in-process request running in this task came through.
123 : static VIEW: std::sync::Arc<str>;
124 : }
125 :
126 : /// Runs `f` as a request through the view `id`: every [`Subject`] built in
127 : /// it carries the [`VIEW_PAIR`]. The view surfaces' operation layer is the
128 : /// one caller; a request off the socket never runs inside it.
129 12 : pub async fn through_view<F: Future>(id: &str, f: F) -> F::Output {
130 12 : VIEW.scope(std::sync::Arc::from(id), f).await
131 12 : }
132 :
133 : impl Subject {
134 : /// The view the request came through, if it came through one.
135 : #[must_use]
136 16 : pub fn view(&self) -> Option<&str> {
137 16 : self.headers
138 16 : .iter()
139 16 : .find(|(name, _)| name == VIEW_PAIR)
140 16 : .map(|(_, id)| id.as_str())
141 16 : }
142 : }
143 :
144 : impl std::fmt::Debug for Subject {
145 4 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
146 4 : f.debug_struct("Subject")
147 4 : .field("tenant", &self.tenant)
148 4 : .field(
149 4 : "headers",
150 4 : &self.headers.iter().map(|(k, _)| k).collect::<Vec<_>>(),
151 : )
152 4 : .finish()
153 4 : }
154 : }
155 :
156 : /// What is being asked for, after negotiation and expansion: the request
157 : /// the broker is about to run, in the terms the operation itself uses.
158 : /// Every name is expanded, so an engine writes its rules against IRIs and
159 : /// not against whatever short name the caller's `@context` happened to use.
160 : pub struct Operation<'a> {
161 : /// The CIM 009 clause of the operation, e.g. `"5.6.1"`.
162 : pub clause: &'static str,
163 : /// The Entity identifiers the operation names, as the handler holds
164 : /// them; empty for a query that names none.
165 : pub ids: &'a [&'a str],
166 : pub types: &'a [String],
167 : pub attrs: &'a [String],
168 : pub q: Option<&'a QNode>,
169 : pub scope_q: Option<&'a str>,
170 : pub geo: Option<&'a GeoQuery>,
171 : /// The request body of a write, expanded. `None` for a read.
172 : pub body: Option<&'a Value>,
173 : /// What a write by id lands on, as stored: `None`
174 : /// for a read, a create, or an id the tenant does not hold.
175 : pub stored: Option<&'a StoredImage>,
176 : }
177 :
178 : /// The stored Entity a write by id changes, as an engine judges it
179 : /// (operation matrix item 2): what the Entity is,
180 : /// not what the request says it is. Names only, expanded; never a value.
181 : #[derive(Debug, Clone, Default, PartialEq, Eq)]
182 : pub struct StoredImage {
183 : pub types: Vec<String>,
184 : pub scope: Vec<String>,
185 : pub attribute_names: Vec<String>,
186 : }
187 :
188 : impl StoredImage {
189 : /// The image of one stored, expanded Entity document.
190 286 : pub fn of(doc: &Value) -> Self {
191 332 : let strings = |v: Option<&Value>| match v {
192 8 : Some(Value::String(s)) => vec![s.clone()],
193 324 : Some(Value::Array(a)) => a
194 324 : .iter()
195 324 : .filter_map(Value::as_str)
196 324 : .map(str::to_owned)
197 324 : .collect(),
198 240 : _ => Vec::new(),
199 572 : };
200 : Self {
201 286 : types: strings(doc.get("type")),
202 286 : scope: strings(doc.get("scope")),
203 286 : attribute_names: doc
204 286 : .as_object()
205 286 : .map(|o| {
206 286 : o.keys()
207 1648 : .filter(|k| {
208 1648 : !k.starts_with("__")
209 1644 : && !k.starts_with('@')
210 1162 : && !matches!(
211 1640 : k.as_str(),
212 1640 : "id" | "type"
213 1076 : | "scope"
214 1030 : | "createdAt"
215 756 : | "modifiedAt"
216 482 : | "deletedAt"
217 : )
218 1648 : })
219 286 : .cloned()
220 286 : .collect()
221 286 : })
222 286 : .unwrap_or_default(),
223 : }
224 286 : }
225 : }
226 :
227 : /// Shape and counts, never the payload: an `Operation` carries the
228 : /// caller's data, and a log line is not where it belongs.
229 : impl std::fmt::Debug for Operation<'_> {
230 8 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231 8 : f.debug_struct("Operation")
232 8 : .field("clause", &self.clause)
233 8 : .field("ids", &self.ids.len())
234 8 : .field("types", &self.types.len())
235 8 : .field("attrs", &self.attrs.len())
236 8 : .field("q", &self.q.is_some())
237 8 : .field("scope_q", &self.scope_q.is_some())
238 8 : .field("geo", &self.geo.is_some())
239 8 : .field("body", &self.body.is_some())
240 8 : .field("stored", &self.stored.is_some())
241 8 : .finish()
242 8 : }
243 : }
244 :
245 : impl Operation<'_> {
246 : /// The operation with nothing but its clause: what a handler that
247 : /// carries no ids, types or attributes passes, and the base every other
248 : /// call site fills in with struct-update syntax.
249 3836 : pub const fn new(clause: &'static str) -> Operation<'static> {
250 3836 : Operation {
251 3836 : clause,
252 3836 : ids: &[],
253 3836 : types: &[],
254 3836 : attrs: &[],
255 3836 : q: None,
256 3836 : scope_q: None,
257 3836 : geo: None,
258 3836 : body: None,
259 3836 : stored: None,
260 3836 : }
261 3836 : }
262 : }
263 :
264 : /// What an engine may answer: an operation is allowed, refused, allowed
265 : /// over less than it asked for, or allowed on a condition the stored Entity
266 : /// has to meet.
267 : #[derive(Debug, Clone, PartialEq)]
268 : pub enum Decision {
269 : Allow,
270 : /// Refused, with the engine's own reason. Answered 403 with a
271 : /// ProblemDetails whose type is an Antares URI — Table 6.3.2-1 names no
272 : /// access-denied error, so this is an Antares decision.
273 : Deny(String),
274 : /// Allowed over less. The caller cannot tell a hidden entity from an
275 : /// absent one.
276 : Filter(Filter),
277 : /// Allowed only if the stored Entity the write lands on is one the
278 : /// filter selects: its `q`, `scopeQ`, areas and the types and
279 : /// conditions of its `rules`, judged on stored values, which the
280 : /// [`Operation`] does not carry. The store judges the document it is
281 : /// about to change, inside the write's own read-modify-write, so a
282 : /// concurrent write between this answer and the commit cannot carry the
283 : /// write onto an Entity the filter no longer selects. An Entity it does
284 : /// not select answers as an absent one. Honoured on [`WRITE_BY_ID`], with
285 : /// a filter that selects and projects nothing; anywhere else a deny
286 : /// ([`resolve`]).
287 : AllowIf(Filter),
288 : }
289 :
290 : /// The writes by id a [`Decision::AllowIf`] can condition: each changes or
291 : /// deletes Entities that are already stored, by id, and judges each of them
292 : /// inside its store operation. A create has no stored Entity to judge; Batch
293 : /// Upsert updates the members its condition selects and creates none of the
294 : /// conditioned ones. The temporal writes 5.6.12 to 5.6.15 judge the
295 : /// evolution's latest instances (`Condition::pre_evolution`); 5.6.16 is
296 : /// not one: the temporal seam has no delete that reads the document first.
297 : pub const WRITE_BY_ID: [&str; 16] = [
298 : "5.6.2", "5.6.3", "5.6.4", "5.6.5", "5.6.6", "5.6.17", "5.6.18", "5.6.19", "5.6.8", "5.6.9",
299 : "5.6.10", "5.6.20", "5.6.12", "5.6.13", "5.6.14", "5.6.15",
300 : ];
301 :
302 : /// Why a write conditioned on its stored Entity is refused rather than
303 : /// forwarded: a Context Source's copy is not in this store, so nothing here
304 : /// can judge it.
305 : pub const CONDITION_NOT_FORWARDED: &str =
306 : "a write conditioned on the stored Entity is not forwarded to a Context Source";
307 :
308 : /// How much less. Every member narrows and none widens: the `q`, `scopeQ`
309 : /// and `geo_q` are conjoined with the caller's own, `pick` keeps a subset of
310 : /// the members, `omit` removes some.
311 : #[derive(Debug, Clone, Default, PartialEq)]
312 : pub struct Filter {
313 : /// Conjoined into the query the store runs, on the AST — never on the
314 : /// query string, where the 4.9 precedence of `;` against `|` would have
315 : /// to be re-derived by whoever writes the rule. Boxed, like `geo_q`: a
316 : /// parsed query is half the size of the struct, and every [`Decision`]
317 : /// and [`NotifyDecision`] would carry it inline.
318 : pub q: Option<Box<QNode>>,
319 : /// Conjoined the same way with the request's own `scopeQ` (4.18).
320 : pub scope_q: Option<String>,
321 : /// An area the Entity must lie in (4.10), held beside the request's own
322 : /// geoquery and never in its place: 7.2.5 lets a request carry one, so
323 : /// the store query takes both and every one must hold.
324 : /// A [`GeoQuery`] is only built by `GeoQuery::from_params`, which bounds
325 : /// it like a request's. Boxed: a parsed geometry is large, and every
326 : /// [`Decision`] would carry its size.
327 : pub geo_q: Option<Box<GeoQuery>>,
328 : /// Further areas, every one of which must hold as well: a view and an
329 : /// engine that both narrow by area compose into `geo_q` and these.
330 : /// An engine with one area leaves this empty; read
331 : /// them all through [`Filter::areas`].
332 : pub geo_also: Vec<GeoQuery>,
333 : /// Members removed from every document served.
334 : pub omit: Vec<String>,
335 : /// If non-empty, the only members served beside the document frame.
336 : pub pick: Vec<String>,
337 : /// Attributes allowed per rule: when
338 : /// non-empty, an Entity is served only if at least one [`Rule`] holds
339 : /// for it, and it keeps the union of the `attributes` of the rules that
340 : /// hold, nothing else beside its frame. An Entity no rule holds for is
341 : /// left out of the answer. Applied to every document the answer
342 : /// carries, joined targets and notified Entities included, each by its
343 : /// own rules. It intersects with `pick`: both narrow.
344 : pub rules: Vec<Rule>,
345 : /// Attributes removed per Entity type: an Entity loses every Attribute
346 : /// any of its types omits.
347 : pub omit_by_type: BTreeMap<String, Vec<String>>,
348 : /// Answer [`RESTRICTED_HEADER`]`: true`, so a client can know the
349 : /// answer was narrowed. Narrowing is otherwise silent.
350 : pub restricted: bool,
351 : /// The engine's advisory, sent as `NGSILD-Warning` (TS 104 176 6.3.17):
352 : /// e.g. the Entity types a filter on an Attribute the caller
353 : /// may not filter on left out, so a developer can tell that refusal
354 : /// from an empty answer. The text is the engine's, never an echo of the
355 : /// request, and names only what the caller may already see.
356 : pub warning: Option<Warning>,
357 : /// Windows of `observedAt` the instances of a temporal read (5.7.3,
358 : /// 5.7.4) must fall into, every one of them: the request's own
359 : /// 4.11 window is clamped to them before the store runs, so the store
360 : /// fetches no instance outside them. A window the request does not reach
361 : /// answers nothing, never a refusal that tells where the grant ends. A
362 : /// slice, not a `Vec`: every [`Decision`] carries a `Filter`'s size.
363 : pub temporal: Box<[Window]>,
364 : }
365 :
366 : /// A half-open interval `[from, to)` of `observedAt`; an open end is the
367 : /// beginning or the end of time.
368 : #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
369 : pub struct Window {
370 : /// The earliest instant admitted.
371 : pub from: Option<TimeBound>,
372 : /// The first instant no longer admitted.
373 : pub to: Option<TimeBound>,
374 : }
375 :
376 : /// One end of a [`Window`]: a fixed instant, or a duration before the
377 : /// moment the request is decided, so "the last 24 hours" slides with now.
378 : #[derive(Debug, Clone, Copy, PartialEq, Eq)]
379 : pub enum TimeBound {
380 : /// This instant.
381 : At(chrono::DateTime<chrono::Utc>),
382 : /// This long before now.
383 : Ago(chrono::Duration),
384 : }
385 :
386 : impl TimeBound {
387 2648 : fn at(self, now: chrono::DateTime<chrono::Utc>) -> chrono::DateTime<chrono::Utc> {
388 2648 : match self {
389 2636 : Self::At(t) => t,
390 12 : Self::Ago(d) => now
391 12 : .checked_sub_signed(d)
392 12 : .unwrap_or(chrono::DateTime::<chrono::Utc>::MIN_UTC),
393 : }
394 2648 : }
395 :
396 : /// The bound at `now` in milliseconds, the precision 4.11 accepts: a
397 : /// start (`up`) rounds up and an end down, so rounding only narrows.
398 2648 : fn render(self, now: chrono::DateTime<chrono::Utc>, up: bool) -> String {
399 2648 : let t = self.at(now);
400 2648 : let sub = i64::from(t.timestamp_subsec_nanos() % 1_000_000);
401 2648 : let t = if up && sub > 0 {
402 20 : t.checked_add_signed(chrono::Duration::nanoseconds(
403 20 : 1_000_000_i64.saturating_sub(sub),
404 : ))
405 20 : .unwrap_or(t)
406 : } else {
407 2628 : t
408 : };
409 2648 : t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
410 2648 : }
411 : }
412 :
413 : impl Window {
414 : /// Whether an instance observed at `observed_at` falls inside, the
415 : /// bounds resolved against `now`; 4.11's half-open comparison.
416 : #[must_use]
417 2436 : pub fn admits(&self, observed_at: &str, now: chrono::DateTime<chrono::Utc>) -> bool {
418 2436 : let t = antares_model::dt_key(observed_at);
419 2436 : self.from
420 2436 : .is_none_or(|b| t >= antares_model::dt_key(&b.render(now, true)))
421 2374 : && self
422 2374 : .to
423 2374 : .is_none_or(|b| t < antares_model::dt_key(&b.render(now, false)))
424 2436 : }
425 :
426 : /// Whether every instance observed at or after `start` is past this
427 : /// window's start, the bound resolved against `now`.
428 : #[must_use]
429 82 : pub fn opens_by(&self, start: &str, now: chrono::DateTime<chrono::Utc>) -> bool {
430 82 : self.from.is_none_or(|b| {
431 82 : antares_model::dt_key(start) >= antares_model::dt_key(&b.render(now, true))
432 82 : })
433 82 : }
434 : }
435 :
436 : /// The text of a [`Filter::warning`]: printable ASCII on one line, at most
437 : /// [`Warning::MAX_LEN`] bytes, no `"` or `\`, so it travels as one quoted
438 : /// `warn-text` of an `NGSILD-Warning` value. Composed filters keep each
439 : /// side's text, at most [`Warning::MAX_TEXTS`].
440 : #[derive(Debug, Clone, PartialEq, Eq)]
441 : pub struct Warning(Box<[String]>); // a slice: every Decision carries a Filter's size
442 :
443 : impl Warning {
444 : /// Bytes in one text.
445 : pub const MAX_LEN: usize = 256;
446 : /// Texts a composed filter keeps.
447 : pub const MAX_TEXTS: usize = 4;
448 :
449 : /// One advisory.
450 : ///
451 : /// # Errors
452 : /// When `text` is empty, longer than [`Warning::MAX_LEN`], or holds a
453 : /// character outside printable ASCII (a line break, a control
454 : /// character), a `"` or a `\\`: it could not travel as one header line.
455 84 : pub fn new(text: &str) -> Result<Self, String> {
456 84 : if text.is_empty() || text.len() > Self::MAX_LEN {
457 8 : return Err(format!(
458 8 : "a policy warning holds 1 to {} bytes",
459 8 : Self::MAX_LEN
460 8 : ));
461 76 : }
462 76 : if !text
463 76 : .bytes()
464 1352 : .all(|b| (0x20..=0x7e).contains(&b) && b != b'"' && b != b'\\')
465 : {
466 24 : return Err(
467 24 : "a policy warning is printable ASCII on one line, without '\"' or '\\\\'".into(),
468 24 : );
469 52 : }
470 52 : Ok(Self(Box::new([text.to_owned()])))
471 84 : }
472 :
473 : /// The texts, in the order the filters were composed.
474 : #[must_use]
475 12 : pub fn texts(&self) -> &[String] {
476 12 : &self.0
477 12 : }
478 :
479 : /// `a` and `b` as one: each text once, `a`'s first.
480 : #[must_use]
481 56 : pub fn join(a: Option<&Self>, b: Option<Self>) -> Option<Self> {
482 56 : match (a, b) {
483 44 : (Some(a), Some(b)) => {
484 44 : let mut texts = a.0.to_vec();
485 44 : for t in b.0 {
486 44 : if !texts.contains(&t) {
487 40 : texts.push(t);
488 40 : }
489 : }
490 44 : texts.truncate(Self::MAX_TEXTS);
491 44 : Some(Self(texts.into()))
492 : }
493 4 : (Some(a), None) => Some(a.clone()),
494 8 : (None, b) => b,
495 : }
496 56 : }
497 : }
498 :
499 : /// One allowlist of a [`Filter`]: the Entities of
500 : /// `types` for which the condition holds show `attributes`. The condition
501 : /// (`q`, `scope_q`, `geo_q`, each optional) is the policy's: it is read on
502 : /// the stored Entity, which may carry Attributes the caller cannot see.
503 : #[derive(Debug, Clone, Default, PartialEq)]
504 : pub struct Rule {
505 : /// Type names, at least one; a rule naming none holds for nothing.
506 : pub types: Vec<String>,
507 : /// On the Entity's own members. A linked path (4.9
508 : /// `LinkedEntityRelation`) never holds: the visible set of one Entity
509 : /// must not depend on another Entity the caller may not read.
510 : pub q: Option<Box<QNode>>,
511 : pub scope_q: Option<String>,
512 : pub geo_q: Option<Box<GeoQuery>>,
513 : /// Further areas the Entity must lie in, as [`Filter::geo_also`].
514 : pub geo_also: Vec<GeoQuery>,
515 : /// Entity ids the rule holds for. With `id_patterns`, an
516 : /// Entity is selected when its id is one of these or matches one of
517 : /// them; both empty select every id.
518 : pub ids: Vec<String>,
519 : /// Regular expressions an Entity id may match instead, each anchored
520 : /// at both ends. One that does not compile, or compiles past
521 : /// `antares_ql::regex`'s bound, matches nothing (fail closed).
522 : pub id_patterns: Vec<String>,
523 : /// The Attributes an Entity this rule holds for shows.
524 : pub attributes: Vec<String>,
525 : /// The rule shows every Attribute, and `attributes` is not read
526 : /// (amended): an engine's grant over the
527 : /// whole Entity. A view never sets it.
528 : pub every_attribute: bool,
529 : /// Windows of `observedAt`: in a temporal read (5.7.3, 5.7.4)
530 : /// an instance of these `attributes` is shown through this rule only if
531 : /// it falls into every one. A current-state read shows what the rule
532 : /// shows, as the gateway's grant windows did.
533 : pub temporal: Vec<Window>,
534 : }
535 :
536 : impl Rule {
537 : /// Every area the Entity must lie in: `geo_q` and `geo_also`.
538 11700 : pub fn areas(&self) -> impl Iterator<Item = &GeoQuery> {
539 11700 : self.geo_q.as_deref().into_iter().chain(&self.geo_also)
540 11700 : }
541 :
542 : /// Whether the rule names `doc` by one of its types and, when it
543 : /// narrows ids, by its id. Neither changes over an Entity's history,
544 : /// so a temporal instance is selected the same way (unlike the
545 : /// condition, [`Rule::is_conditional`]).
546 12642 : pub fn selects(&self, doc: &Value) -> bool {
547 : // `entity_types` read in place: this runs per rule per Entity of a
548 : // page, and the list it would allocate is read once.
549 13346 : let named = |t: &str| self.types.iter().any(|r| r == t);
550 12642 : let typed = match doc.get("type") {
551 48 : Some(Value::String(t)) => named(t),
552 12594 : Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).any(named),
553 0 : _ => false,
554 : };
555 12642 : typed
556 8334 : && match doc.get("id").and_then(Value::as_str) {
557 8326 : Some(id) => self.selects_id(id),
558 8 : None => self.ids.is_empty() && self.id_patterns.is_empty(),
559 : }
560 12642 : }
561 :
562 : /// Whether the rule's `ids` and `id_patterns` select `id`.
563 8326 : pub fn selects_id(&self, id: &str) -> bool {
564 8326 : (self.ids.is_empty() && self.id_patterns.is_empty())
565 442 : || self.ids.iter().any(|i| i == id)
566 396 : || self.id_patterns.iter().any(|p| {
567 362 : antares_ql::regex::compile(&format!("^(?:{p})$")).is_ok_and(|re| re.is_match(id))
568 362 : })
569 8326 : }
570 :
571 : /// Whether the rule has a condition beside its types.
572 3924 : pub fn is_conditional(&self) -> bool {
573 3924 : self.q.is_some() || self.scope_q.is_some() || self.areas().next().is_some()
574 3924 : }
575 :
576 : /// Whether the rule holds for exactly the Entities of its types: no
577 : /// condition, no ids, no id patterns (what [`Self::holds`] reads beside
578 : /// the types).
579 432 : pub fn admits_by_type_alone(&self) -> bool {
580 432 : !self.is_conditional() && self.ids.is_empty() && self.id_patterns.is_empty()
581 432 : }
582 :
583 : /// Whether the rule holds for `doc`, an Entity in the form `ctx`
584 : /// expands names to (the broker's internal form: expanded keys, each
585 : /// Attribute an array of instances).
586 12492 : pub fn holds(&self, doc: &Value, ctx: &antares_jsonld::Context) -> bool {
587 12492 : self.selects(doc)
588 8160 : && self
589 8160 : .scope_q
590 8160 : .as_deref()
591 8160 : .is_none_or(|s| antares_ql::scope::scope_matches(s, doc))
592 8160 : && self.areas().all(|g| g.matches(doc, ctx))
593 7952 : && self.q.as_deref().is_none_or(|q| {
594 2234 : !q.has_linked_paths() && antares_ql::eval::eval_q(q, doc, ctx, &|_| None)
595 2234 : })
596 12492 : }
597 :
598 : /// The Attributes the condition reads, expanded through `ctx`: the head
599 : /// of every `q` term and the geoquery's GeoProperty.
600 82 : pub fn reads(&self, ctx: &antares_jsonld::Context) -> Vec<String> {
601 82 : let mut out: Vec<String> = self
602 82 : .q
603 82 : .as_deref()
604 82 : .map(|q| {
605 14 : q.attribute_paths()
606 14 : .into_iter()
607 14 : .filter_map(|p| p.split(['.', '[', '{']).next())
608 14 : .map(|h| ctx.expand_key(h))
609 14 : .collect()
610 14 : })
611 82 : .unwrap_or_default();
612 82 : for g in self.areas() {
613 14 : out.push(if g.geoproperty.is_empty() {
614 14 : antares_ql::geo::LOCATION_IRI.to_owned()
615 : } else {
616 0 : ctx.expand_key(&g.geoproperty)
617 : });
618 : }
619 82 : out.sort();
620 82 : out.dedup();
621 82 : out
622 82 : }
623 :
624 : /// the narrowing rules disclosure: the Attributes the condition reads that
625 : /// the rule does not show. Seeing the rule's Attributes tells the
626 : /// caller the condition held, so these are what the visible set
627 : /// discloses; the view loader reports them.
628 12 : pub fn discloses(&self, ctx: &antares_jsonld::Context) -> Vec<String> {
629 12 : if self.every_attribute {
630 4 : return Vec::new();
631 8 : }
632 8 : self.reads(ctx)
633 8 : .into_iter()
634 16 : .filter(|r| !self.attributes.iter().any(|a| ctx.expand_key(a) == *r))
635 8 : .collect()
636 12 : }
637 : }
638 :
639 : /// The answer of a broker without an engine: a [`Filter`] that narrows
640 : /// nothing, one shared instance so asking costs no allocation.
641 50250 : pub fn unnarrowed() -> std::sync::Arc<Filter> {
642 : static NOTHING: LazyLock<std::sync::Arc<Filter>> =
643 46 : LazyLock::new(|| std::sync::Arc::new(Filter::default()));
644 50250 : NOTHING.clone()
645 50250 : }
646 :
647 : /// What a [`Filter`]'s rules let one Entity show: `only` (when set) is the
648 : /// union of the matched rules' Attributes, `omitted` what its types omit.
649 : #[derive(Debug, Clone, Default, PartialEq)]
650 : ///
651 : /// The names borrow from the rules that decided them, so judging an Entity
652 : /// copies no Attribute name; [`Shown::into_owned`] and
653 : /// [`Shown::map_names`] give a set that outlives the rules.
654 : pub struct Shown<'a> {
655 : pub only: Option<std::collections::BTreeSet<Cow<'a, str>>>,
656 : pub omitted: std::collections::BTreeSet<Cow<'a, str>>,
657 : }
658 :
659 : impl Shown<'_> {
660 : /// Nothing beside the frame: what a projection falls back to for an
661 : /// Entity no rule holds for, should one reach it.
662 4 : pub fn nothing() -> Self {
663 4 : Shown {
664 4 : only: Some(Default::default()),
665 4 : omitted: Default::default(),
666 4 : }
667 4 : }
668 :
669 36580 : pub fn allows(&self, k: &str) -> bool {
670 36580 : self.only.as_ref().is_none_or(|o| o.contains(k)) && !self.omitted.contains(k)
671 36580 : }
672 :
673 : /// The same set with every name mapped, for a document in another form.
674 8 : pub fn map_names(&self, f: impl Fn(&str) -> String) -> Shown<'static> {
675 : Shown {
676 8 : only: self
677 8 : .only
678 8 : .as_ref()
679 8 : .map(|o| o.iter().map(|n| Cow::Owned(f(n))).collect()),
680 8 : omitted: self.omitted.iter().map(|n| Cow::Owned(f(n))).collect(),
681 : }
682 8 : }
683 :
684 : /// The same set, owning its names.
685 4 : pub fn into_owned(self) -> Shown<'static> {
686 8 : let own = |n: Cow<'_, str>| Cow::Owned(n.into_owned());
687 : Shown {
688 4 : only: self.only.map(|o| o.into_iter().map(own).collect()),
689 4 : omitted: self.omitted.into_iter().map(own).collect(),
690 : }
691 4 : }
692 : }
693 :
694 : /// The types of an Entity document: its `type` member, a string or an
695 : /// array of strings (5.2.4), in whatever form the document is in.
696 814 : pub fn entity_types(doc: &Value) -> Vec<&str> {
697 814 : match doc.get("type") {
698 8 : Some(Value::String(t)) => vec![t.as_str()],
699 708 : Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).collect(),
700 98 : _ => Vec::new(),
701 : }
702 814 : }
703 :
704 : /// What identifies the document rather than describing it: 5.2.4 makes
705 : /// `id` and `type` mandatory members of an Entity, and an answer without
706 : /// them is not an Entity at all. A projection never removes these.
707 : const FRAME: [&str; 3] = ["id", "type", "@context"];
708 :
709 : impl Filter {
710 : /// Every area the Entity must lie in: `geo_q` and `geo_also`.
711 28213 : pub fn areas(&self) -> impl Iterator<Item = &GeoQuery> {
712 28213 : self.geo_q.as_deref().into_iter().chain(&self.geo_also)
713 28213 : }
714 :
715 : /// Apply `pick`/`omit` and the rules to one document by member name.
716 : /// This is the reference semantics `run_policy_contract` holds an
717 : /// engine's answer against; a served document is projected through the
718 : /// request's own 5.5.2 representation instead
719 : /// (`repr::narrow_projection`), where the names are expanded against
720 : /// the request `@context` first. Rule conditions are read with the core
721 : /// `@context`; a document no rule holds for keeps its frame here, while
722 : /// the served paths leave it out.
723 42 : pub fn project(&self, doc: &mut Value) {
724 42 : let shown = self.has_rules().then(|| {
725 4 : self.shown(doc, &CORE_CONTEXT)
726 4 : .unwrap_or_else(Shown::nothing)
727 4 : });
728 42 : let Some(obj) = doc.as_object_mut() else {
729 4 : return;
730 : };
731 38 : if !self.pick.is_empty() {
732 36 : obj.retain(|k, _| FRAME.contains(&k.as_str()) || self.pick.iter().any(|p| p == k));
733 28 : }
734 38 : for name in &self.omit {
735 28 : if FRAME.contains(&name.as_str()) {
736 8 : continue;
737 20 : }
738 20 : obj.remove(name);
739 : }
740 38 : if let Some(shown) = shown {
741 28 : obj.retain(|k, _| FRAME.contains(&k.as_str()) || shown.allows(k));
742 34 : }
743 42 : }
744 :
745 : /// True when the filter would change nothing, which is how an engine
746 : /// that means "allow" can say so with an empty `Filter`.
747 862 : pub fn is_empty(&self) -> bool {
748 862 : self.q.is_none()
749 830 : && self.scope_q.is_none()
750 830 : && self.areas().next().is_none()
751 826 : && self.omit.is_empty()
752 750 : && self.pick.is_empty()
753 750 : && !self.has_rules()
754 682 : && self.temporal.is_empty()
755 862 : }
756 :
757 : /// Whether the filter removes anything from what it selects: a
758 : /// projection, a per-type omit, a rule's Attribute list or a temporal
759 : /// window. A write condition ([`Decision::AllowIf`]) may not: the write
760 : /// is performed whole or not at all.
761 388 : pub fn projects(&self) -> bool {
762 388 : !self.pick.is_empty()
763 376 : || !self.omit.is_empty()
764 350 : || !self.omit_by_type.is_empty()
765 346 : || !self.temporal.is_empty()
766 342 : || self
767 342 : .rules
768 342 : .iter()
769 342 : .any(|r| !r.attributes.is_empty() || !r.temporal.is_empty())
770 388 : }
771 :
772 : /// Whether the filter carries rules or per-type omits.
773 28319 : pub fn has_rules(&self) -> bool {
774 28319 : !self.rules.is_empty() || !self.omit_by_type.is_empty()
775 28319 : }
776 :
777 : /// Whether a rule holds for `doc` (every document without rules): what
778 : /// [`Self::shown`] answers `Some` for, without building the set.
779 1716 : pub fn admits(&self, doc: &Value, ctx: &antares_jsonld::Context) -> bool {
780 2576 : self.rules.is_empty() || self.rules.iter().any(|r| r.holds(doc, ctx))
781 1716 : }
782 :
783 : /// what `doc` may show, or `None` when `rules` is
784 : /// non-empty and none holds for it (the Entity is out). An Entity sees
785 : /// the union of the Attributes of the rules that hold for it, less what
786 : /// any of its types omits. Names compare as written: the caller brings
787 : /// rules and document into one form first (`repr::type_rules` expands
788 : /// both), and `ctx` is the one the conditions' names expand through.
789 5680 : pub fn shown(&self, doc: &Value, ctx: &antares_jsonld::Context) -> Option<Shown<'_>> {
790 5680 : let only = if self.rules.is_empty() {
791 18 : None
792 : } else {
793 9670 : let mut held = self.rules.iter().filter(|r| r.holds(doc, ctx)).peekable();
794 5662 : held.peek()?;
795 4788 : let held: Vec<&Rule> = held.collect();
796 4788 : if held.iter().any(|r| r.every_attribute) {
797 2 : None
798 : } else {
799 : Some(
800 4786 : held.iter()
801 10056 : .flat_map(|r| r.attributes.iter().map(|a| Cow::Borrowed(a.as_str())))
802 4786 : .collect(),
803 : )
804 : }
805 : };
806 4806 : if self.omit_by_type.is_empty() {
807 4784 : return Some(Shown {
808 4784 : only,
809 4784 : omitted: Default::default(),
810 4784 : });
811 22 : }
812 22 : let omitted = entity_types(doc)
813 22 : .into_iter()
814 32 : .filter_map(|t| self.omit_by_type.get(t))
815 22 : .flatten()
816 22 : .map(|a| Cow::Borrowed(a.as_str()))
817 22 : .collect();
818 22 : Some(Shown { only, omitted })
819 5680 : }
820 :
821 : /// the narrowing rules per rule (K22): a write is allowed only when some
822 : /// rule holds for its post-image and every Attribute the write names is
823 : /// shown by a rule that holds, none of which its types omit: the write
824 : /// sets what the Entity then shows, the union of its rules. One rule
825 : /// need not show them all, so a type whose list a view splits over
826 : /// several rules is written whole. The check a view's engine
827 : /// runs before it allows a write.
828 48 : pub fn admits_write(
829 48 : &self,
830 48 : post_image: &Value,
831 48 : written: &[&str],
832 48 : ctx: &antares_jsonld::Context,
833 48 : ) -> bool {
834 48 : let types = entity_types(post_image);
835 64 : let omitted = |w: &str| {
836 64 : types
837 64 : .iter()
838 64 : .filter_map(|t| self.omit_by_type.get(*t))
839 64 : .any(|o| o.iter().any(|a| a == w))
840 64 : };
841 64 : if written.iter().any(|w| omitted(w)) {
842 4 : return false;
843 44 : }
844 44 : if self.rules.is_empty() {
845 4 : return true;
846 40 : }
847 40 : let holding: Vec<&Rule> = self
848 40 : .rules
849 40 : .iter()
850 92 : .filter(|r| r.holds(post_image, ctx))
851 40 : .collect();
852 40 : !holding.is_empty()
853 32 : && (holding.iter().any(|r| r.every_attribute)
854 28 : || written
855 28 : .iter()
856 68 : .all(|w| holding.iter().any(|r| r.attributes.iter().any(|a| a == w))))
857 48 : }
858 :
859 : /// Say the answer was narrowed, and add the engine's warning, when the
860 : /// engine asked for either. Narrowing is otherwise silent: a caller
861 : /// cannot tell an Entity it may not see from one that is not there.
862 : /// `alias` is the warn-agent, this broker's host alias.
863 6358 : pub fn mark(&self, alias: &str, headers: &mut HeaderMap) {
864 6358 : if self.restricted {
865 30 : headers.insert(
866 30 : RESTRICTED_HEADER,
867 30 : axum::http::HeaderValue::from_static("true"),
868 30 : );
869 6328 : }
870 6358 : for text in self.warning.iter().flat_map(Warning::texts) {
871 : // 199 Miscellaneous warning (RFC 7234 5.5), as the gateway's
872 : // dropped-types warning was.
873 4 : let value = crate::federation::warning(199, alias, text);
874 4 : if let Ok(v) = axum::http::HeaderValue::from_str(&value) {
875 4 : headers.append("NGSILD-Warning", v);
876 4 : }
877 : }
878 6358 : }
879 :
880 : /// The request's temporal parameters with its 4.11 window clamped to
881 : /// [`Filter::temporal`], resolved against now, for every reader
882 : /// after the gate: the store's instance range, the forward, the links.
883 : /// `required` is the operation's (5.7.4 demands a window, 5.7.3 does
884 : /// not), so a request that lacked one is still refused. The windows are
885 : /// on `observedAt`: a request on another timeproperty is answered
886 : /// nothing, since the store applies one range on one timeproperty.
887 : ///
888 : /// # Errors
889 : /// The request's own 4.11 window is invalid.
890 2244 : pub fn narrow_window(
891 2244 : &self,
892 2244 : params: &std::collections::HashMap<String, String>,
893 2244 : required: bool,
894 2244 : ) -> Result<std::collections::HashMap<String, String>, antares_model::NgsiError> {
895 2244 : let mut out = params.clone();
896 2244 : if self.temporal.is_empty() {
897 2180 : return Ok(out);
898 64 : }
899 64 : let asked = crate::temporalq::TemporalQ::from_params(params, required)?;
900 148 : let key = |s: &String| antares_model::dt_key(s);
901 56 : let (mut from, mut to) = match &asked {
902 34 : Some(tq) if tq.timeproperty != "observedAt" => {
903 6 : let epoch = "1970-01-01T00:00:00Z".to_owned();
904 6 : (Some(epoch.clone()), Some(epoch))
905 : }
906 28 : Some(tq) => match tq.timerel.as_str() {
907 28 : "before" => (None, Some(tq.time_at.clone())),
908 18 : "after" => (Some(tq.time_at.clone()), None),
909 4 : "between" => (Some(tq.time_at.clone()), tq.end_time_at.clone()),
910 0 : _ => (None, None),
911 : },
912 22 : None => (None, None),
913 : };
914 56 : let now = chrono::Utc::now();
915 66 : for w in &self.temporal {
916 66 : if let Some(b) = w.from.map(|b| b.render(now, true)) {
917 44 : from = Some(match from {
918 20 : Some(f) if key(&f) >= key(&b) => f,
919 40 : _ => b,
920 : });
921 22 : }
922 66 : if let Some(b) = w.to.map(|b| b.render(now, false)) {
923 38 : to = Some(match to {
924 18 : Some(t) if key(&t) <= key(&b) => t,
925 24 : _ => b,
926 : });
927 28 : }
928 : }
929 : // `between` an instant and itself admits nothing: the answer is
930 : // empty, as for a window over a period with no data
931 56 : if let (Some(f), Some(t)) = (&from, &to) {
932 36 : if key(f) >= key(t) {
933 18 : to = Some(f.clone());
934 18 : }
935 20 : }
936 56 : let (timerel, time_at, end) = match (from, to) {
937 36 : (Some(f), Some(t)) => ("between", f, Some(t)),
938 12 : (Some(f), None) => ("after", f, None),
939 4 : (None, Some(t)) => ("before", t, None),
940 4 : (None, None) => return Ok(out),
941 : };
942 52 : out.insert("timerel".into(), timerel.into());
943 52 : out.insert("timeAt".into(), time_at);
944 52 : match end {
945 36 : Some(e) => out.insert("endTimeAt".into(), e),
946 16 : None => out.remove("endTimeAt"),
947 : };
948 52 : Ok(out)
949 2244 : }
950 :
951 : /// The request's own query parameters, narrowed by this decision.
952 : ///
953 : /// The `q` conjunction is made on the AST and rendered back once
954 : /// (`antares_ql`'s renderer parenthesises an `Or` inside an `And`, so
955 : /// 4.9's `;`-over-`|` precedence is the renderer's problem and not the
956 : /// rule writer's). Every consumer below reads the narrowed parameters:
957 : /// the store push-down, the local re-check that 5.7.2.4 runs over
958 : /// merged results, and the query the request is forwarded with.
959 : ///
960 : /// A `scopeQ` narrowing is set when the request carries none, and
961 : /// intersected with the request's own when it carries one. 4.19's `and`
962 : /// is over independent per-pattern predicates, so it distributes over
963 : /// the `,`/`|` disjunction and the intersection is itself a Scope
964 : /// Query — `antares_ql::scope::intersect_scope_q` writes it. Where the
965 : /// product cannot be written down the seam answers the only way that is
966 : /// not wider than the engine decided: it refuses.
967 : ///
968 : /// `geo_q` stays out of the parameters: they hold one geoquery (7.2.5)
969 : /// and it is the request's. The narrowing's area reaches the store
970 : /// query beside it through `repr::narrowing`.
971 : ///
972 : /// a term of the request's own `q` on an Attribute
973 : /// this `pick`/`omit` hides reads as a term on an Attribute the Entity
974 : /// does not have (see [`Filter::hide_terms`]). `ctx` is the request's,
975 : /// the one its `q` terms expand through.
976 3786 : pub fn narrow_params(
977 3786 : &self,
978 3786 : params: &std::collections::HashMap<String, String>,
979 3786 : ctx: &antares_jsonld::Context,
980 3786 : ) -> Result<std::collections::HashMap<String, String>, Denied> {
981 3786 : let mut out = params.clone();
982 : // the request's own `q` is parsed and refused long before a filter
983 : // reaches it; an unparsable one here is a caller that narrowed
984 : // something it never validated
985 3786 : let own = match out.get("q").map(|q| antares_ql::parse_q(q)).transpose() {
986 3786 : Ok(own) => own.map(|own| self.hide_terms(own, ctx)),
987 0 : Err(_) => return Err(Denied(ENGINE_FAILED.to_owned())),
988 : };
989 : // threat V23: the policy part is held to its share of the narrowed
990 : // caps on its own, so whether it fits never depends on the caller's
991 : // `q` (see `antares_ql::parse_narrowed_q`)
992 3786 : if self
993 3786 : .q
994 3786 : .as_deref()
995 3786 : .is_some_and(|q| !antares_ql::fits_policy_q(q))
996 : {
997 6 : return Err(Denied(ENGINE_FAILED.to_owned()));
998 3780 : }
999 3780 : let narrowed = match (own, &self.q) {
1000 34 : (Some(own), Some(extra)) => Some(QNode::And(vec![own, (**extra).clone()])),
1001 3746 : (own, extra) => own.or_else(|| extra.as_deref().cloned()),
1002 : };
1003 3780 : if let Some(narrowed) = narrowed {
1004 776 : out.insert("q".to_owned(), narrowed.to_string());
1005 3004 : }
1006 3780 : if let Some(scope) = &self.scope_q {
1007 22 : let narrowed = match out.get("scopeQ") {
1008 14 : None => scope.clone(),
1009 8 : Some(own) => antares_ql::scope::intersect_scope_q(own, scope)
1010 8 : .ok_or_else(|| Denied(SCOPE_NOT_NARROWABLE.to_owned()))?,
1011 : };
1012 20 : out.insert("scopeQ".to_owned(), narrowed);
1013 3758 : }
1014 3778 : Ok(out)
1015 3786 : }
1016 : }
1017 :
1018 : impl Filter {
1019 : /// Whether this `pick`/`omit` hides the Attribute a q term names as
1020 : /// `written`, `iri` once expanded. A pick narrows Attributes only: the
1021 : /// system members it never names stay readable, as in the projection.
1022 5768 : fn hides(&self, written: &str, iri: &str) -> bool {
1023 5780 : let named = |names: &[String]| {
1024 5780 : names
1025 5780 : .iter()
1026 5780 : .any(|n| crate::repr::policy_context().expand_key(n) == iri)
1027 5780 : };
1028 5768 : named(&self.omit)
1029 3096 : || (!self.pick.is_empty()
1030 12 : && antares_ql::is_non_system_attr(written)
1031 12 : && !named(&self.pick))
1032 5768 : }
1033 :
1034 : /// the narrowing rules: a term on an Attribute this narrowing
1035 : /// hides reads as a term on one the Entity does not have — a
1036 : /// comparison or `attr` is false, `!attr` is true — so the hidden value
1037 : /// is never compared, on the store path or in memory. The q grammar
1038 : /// has no constants, so false is written `a;!a` and true `a|!a`, on
1039 : /// the hidden name itself: both hold whatever the Entity carries. The
1040 : /// path's top Attribute decides, so a sub-attribute or a hop through a
1041 : /// hidden Relationship is as hidden as the Attribute. Per-type rules
1042 : /// are not read here: their q is read on what the rules show
1043 : /// (`repr::client_q`).
1044 6484 : pub fn hide_terms(&self, node: QNode, ctx: &antares_jsonld::Context) -> QNode {
1045 6484 : if self.omit.is_empty() && self.pick.is_empty() {
1046 704 : return node;
1047 5780 : }
1048 5780 : let constant = |top: &str, holds: bool| {
1049 2670 : let is = |negated| QNode::Exists {
1050 5340 : path: antares_ql::QPath::dotted(vec![top.to_owned()]),
1051 5340 : negated,
1052 5340 : };
1053 2670 : let pair = vec![is(false), is(true)];
1054 2670 : if holds {
1055 6 : QNode::Or(pair)
1056 : } else {
1057 2664 : QNode::And(pair)
1058 : }
1059 2670 : };
1060 5780 : let hidden = |path: &antares_ql::QPath| {
1061 5752 : path.top()
1062 5752 : .filter(|top| self.hides(top, &ctx.expand_key(top)))
1063 5752 : .map(str::to_owned)
1064 5752 : };
1065 5780 : match node {
1066 2 : QNode::And(nodes) => {
1067 4 : QNode::And(nodes.into_iter().map(|n| self.hide_terms(n, ctx)).collect())
1068 : }
1069 26 : QNode::Or(nodes) => {
1070 5726 : QNode::Or(nodes.into_iter().map(|n| self.hide_terms(n, ctx)).collect())
1071 : }
1072 5744 : QNode::Cmp { path, op, value } => match hidden(&path) {
1073 2662 : Some(top) => constant(&top, false),
1074 3082 : None => QNode::Cmp { path, op, value },
1075 : },
1076 8 : QNode::Exists { path, negated } => match hidden(&path) {
1077 8 : Some(top) => constant(&top, negated),
1078 0 : None => QNode::Exists { path, negated },
1079 : },
1080 : }
1081 6484 : }
1082 : }
1083 :
1084 : /// The response header a `Filter { restricted: true }` adds. It is not in
1085 : /// the `NGSILD-` namespace: that prefix is ETSI's, carries the headers
1086 : /// clause 6.3 defines (`NGSILD-Tenant`, `NGSILD-EntityMap`,
1087 : /// `NGSILD-Results-Count`, `NGSILD-Warning`), and a broker-invented header
1088 : /// under it would collide with whatever a later version puts there — the
1089 : /// same reason a refusal answers `urn:antares:error:AccessDenied` rather
1090 : /// than an invented `uri.etsi.org` type.
1091 : pub const RESTRICTED_HEADER: &str = "Antares-Results-Restricted";
1092 :
1093 : /// The refusal a `scopeQ` narrowing gets when its intersection with the
1094 : /// request's own cannot be written as a Scope Query — one side selects
1095 : /// nothing, or the distributed product is too large to express (see
1096 : /// [`Filter::narrow_params`]).
1097 : pub const SCOPE_NOT_NARROWABLE: &str =
1098 : "the request's own scope query cannot be narrowed by the policy engine";
1099 :
1100 : /// What an engine may answer about one notification, for one subscription.
1101 : #[derive(Debug, Clone, PartialEq)]
1102 : pub enum NotifyDecision {
1103 : Deliver,
1104 : /// Send it, narrowed: the same `pick`/`omit` projection the query path
1105 : /// applies, over the entities of `data`. A notification `Filter` that
1106 : /// carries `q` or `scopeQ` is refused as a [`NotifyDecision::Drop`]:
1107 : /// the entities were selected by the subscription's own conditions long
1108 : /// before this point, there is nothing left to re-run the query against,
1109 : /// and delivering the notification unfiltered would tell the engine a
1110 : /// narrowing was applied that never was.
1111 : Filter(Filter),
1112 : /// Do not send. 5.8.6 counts this as no attempt at all — the
1113 : /// notification was never sent, so it is neither a success nor a
1114 : /// failure, and `timesSent` does not move.
1115 : Drop,
1116 : }
1117 :
1118 : /// The seam. One implementation ships with the broker; every other one is
1119 : /// an addon crate a deployment builds itself.
1120 : pub trait PolicyEngine: Send + Sync {
1121 : /// The name a deployment selects the engine by, and the name
1122 : /// `/q/health` reports.
1123 : fn name(&self) -> &str;
1124 :
1125 : /// Fires once per request, after negotiation and expansion, before the
1126 : /// operation and before any fan-out (ADR-0014 `on_request`).
1127 : fn decide<'a>(&'a self, subject: &'a Subject, op: &'a Operation<'a>) -> DecisionFuture<'a>;
1128 :
1129 : /// Fires once per batch (5.6.7 to 5.6.10, 5.6.20) with one Operation
1130 : /// per member, so a batch is refused member by member and the allowed
1131 : /// members are applied (207). The answer holds one Decision per
1132 : /// Operation, in order; an answer of any other length denies the whole
1133 : /// batch. The default asks [`decide`](Self::decide) once per member; an
1134 : /// engine that asks a server overrides it to ask once.
1135 74 : fn decide_each<'a>(
1136 74 : &'a self,
1137 74 : subject: &'a Subject,
1138 74 : ops: &'a [Operation<'a>],
1139 74 : ) -> DecisionsFuture<'a> {
1140 74 : Box::pin(async move {
1141 74 : let mut out = Vec::with_capacity(ops.len());
1142 280 : for op in ops {
1143 280 : out.push(self.decide(subject, op).await);
1144 : }
1145 74 : out
1146 74 : })
1147 74 : }
1148 :
1149 : /// The probe a surface lists its operations by (a surface's listing): one
1150 : /// Operation per clause, naming nothing, and each answer says whether
1151 : /// some request of that clause could be allowed for `subject`, never
1152 : /// whether this one is. The default decides the empty Operations as
1153 : /// [`decide_each`](Self::decide_each) would; an engine whose rules need
1154 : /// a type to answer a read overrides it, so a grant that takes the
1155 : /// queried types is offered, and a refusal of every request is not.
1156 4 : fn offers_each<'a>(
1157 4 : &'a self,
1158 4 : subject: &'a Subject,
1159 4 : ops: &'a [Operation<'a>],
1160 4 : ) -> DecisionsFuture<'a> {
1161 4 : self.decide_each(subject, ops)
1162 4 : }
1163 :
1164 : /// Fires once per notification document per subscription, before the
1165 : /// egress check and the send (ADR-0014 `pre_notify`). Synchronous: it
1166 : /// sits inside the delivery the broker is about to make, and an engine
1167 : /// that has to ask a server for every notification is a design the seam
1168 : /// declines to make easy.
1169 : fn pre_notify(
1170 : &self,
1171 : subject: &Subject,
1172 : sub: &Value,
1173 : notification: &mut Value,
1174 : ) -> NotifyDecision;
1175 :
1176 : /// Whether the engine reads rule documents through
1177 : /// [`load`](Self::load) (ADR-0020 amendment). No by default: `/q/policy`
1178 : /// then answers 404, as it does for a broker without an engine.
1179 4 : fn loads_documents(&self) -> bool {
1180 4 : false
1181 4 : }
1182 :
1183 : /// Take `document` as the rule document of `tenant`, or drop that
1184 : /// Tenant's document when it is `Value::Null`. An `Err` leaves the
1185 : /// engine as it was: the previous document keeps deciding. The reason
1186 : /// goes back to the admin caller who wrote the document and nowhere
1187 : /// else, so it may name parts of it. The default refuses every
1188 : /// document.
1189 : ///
1190 : /// # Errors
1191 : /// Why the engine refuses the document.
1192 4 : fn load(&self, tenant: &str, document: &Value) -> Result<(), String> {
1193 4 : let _ = (tenant, document);
1194 4 : Err(format!(
1195 4 : "the policy engine {:?} reads no rule documents",
1196 4 : self.name()
1197 4 : ))
1198 4 : }
1199 : }
1200 :
1201 : /// The name of the engine the broker ships, which is also the name
1202 : /// `/q/health` reports when no engine is attached at all: the two are the
1203 : /// same decision — allow — and an operator reading it learns what the
1204 : /// broker does, not which object made it happen.
1205 : pub const BUILT_IN_NAME: &str = "allow-all";
1206 :
1207 : /// The engine the broker ships, and the one conformance is asserted
1208 : /// against: it decides nothing, so the broker behaves exactly as it did
1209 : /// before the seam existed.
1210 : ///
1211 : /// A deployment that attaches no engine does not get an instance of this:
1212 : /// it gets no engine at all ([`AppState::policy`](crate::state::AppState)
1213 : /// is `None`), and every gate answers the empty [`Filter`] without building
1214 : /// a [`Subject`], boxing a future or arming the [`TIMEOUT`] timer. The two
1215 : /// paths decide the same thing, which is what
1216 : /// `the_built_in_answer_is_the_bypassed_answer` holds them to.
1217 : pub struct AllowAll;
1218 :
1219 : impl PolicyEngine for AllowAll {
1220 12 : fn name(&self) -> &str {
1221 12 : BUILT_IN_NAME
1222 12 : }
1223 :
1224 224 : fn decide<'a>(&'a self, _subject: &'a Subject, _op: &'a Operation<'a>) -> DecisionFuture<'a> {
1225 224 : Box::pin(std::future::ready(Decision::Allow))
1226 224 : }
1227 :
1228 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
1229 4 : NotifyDecision::Deliver
1230 4 : }
1231 : }
1232 :
1233 : /// A `Filter` only means something where the broker can serve less than it
1234 : /// was asked for, and [`FILTERABLE`] is that list. Everywhere else the
1235 : /// narrowing would be silently dropped: "purge every entity, but only the
1236 : /// ones you may see" is a different operation from the one the client
1237 : /// asked for and would delete less than the 204 claims, and a create writes
1238 : /// its Entity or does not. Those operations take the strict reading: narrow
1239 : /// means refuse. An empty `Filter` asks for nothing and is an allow.
1240 4140 : pub fn resolve(clause: &str, decision: Decision) -> Decision {
1241 1004 : match decision {
1242 332 : Decision::AllowIf(_) if !WRITE_BY_ID.contains(&clause) => Decision::Deny(format!(
1243 30 : "{clause} is not a write by id and cannot be conditioned on the stored Entity"
1244 30 : )),
1245 302 : Decision::AllowIf(f) if f.projects() => {
1246 24 : Decision::Deny("a write condition selects Entities and projects nothing".into())
1247 : }
1248 278 : Decision::AllowIf(f) if Condition::of(&f).is_none() => Decision::Allow,
1249 1032 : Decision::Filter(f) if RECORD_LISTS.contains(&clause) && f.projects() => Decision::Deny(
1250 28 : format!("{clause} lists records whole and a narrowing there cannot project"),
1251 28 : ),
1252 1004 : Decision::Filter(f) if !FILTERABLE.contains(&clause) && !NAME_READS.contains(&clause) => {
1253 94 : if f.is_empty() {
1254 10 : Decision::Allow
1255 84 : } else if WHOLE_TENANT.contains(&clause) {
1256 40 : Decision::Deny(format!(
1257 40 : "{clause} acts on everything the tenant holds and cannot be narrowed"
1258 40 : ))
1259 : } else {
1260 44 : Decision::Deny(format!(
1261 44 : "{clause} is performed whole or not at all and cannot be narrowed"
1262 44 : ))
1263 : }
1264 : }
1265 3898 : other => other,
1266 : }
1267 4140 : }
1268 :
1269 : /// Ask the engine, and fail closed. An engine that panics, or that runs
1270 : /// past [`TIMEOUT`], denies: a seam that waved the request through on its
1271 : /// own failure would turn a broken addon into an open door.
1272 : ///
1273 : /// The panic is caught in place rather than on a spawned task, because the
1274 : /// task boundary would demand `'static` of the operation, which borrows the
1275 : /// request. It is caught TWICE, because an engine has two places to fail:
1276 : /// `decide` returns a boxed future, and a synchronous engine does its whole
1277 : /// decision in the call that builds that future — the reference engine is
1278 : /// `Box::pin(ready(self.judge(..)))` — so guarding only the future guards
1279 : /// the half that does nothing.
1280 : ///
1281 : /// [`TIMEOUT`] can only race the future. Work done before the future exists
1282 : /// holds the executor thread, and no timer inside the same task can
1283 : /// interrupt it; an engine that blocks is a deployment's own bug, and the
1284 : /// bound that catches it is the request timeout in front of the broker.
1285 3528 : pub async fn decide(engine: &dyn PolicyEngine, subject: &Subject, op: &Operation<'_>) -> Decision {
1286 : #[cfg(not(target_arch = "wasm32"))]
1287 : {
1288 : use futures_util::FutureExt as _;
1289 3528 : let built =
1290 3528 : std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| engine.decide(subject, op)));
1291 3528 : let Ok(fut) = built else {
1292 4 : tracing::error!("policy engine {} panicked; denying", engine.name());
1293 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic").increment(1);
1294 4 : return Decision::Deny(ENGINE_FAILED.to_owned());
1295 : };
1296 3524 : let guarded = std::panic::AssertUnwindSafe(fut).catch_unwind();
1297 3524 : match tokio::time::timeout(*TIMEOUT, guarded).await {
1298 3508 : Ok(Ok(d)) => resolve(op.clause, d),
1299 : Ok(Err(_)) => {
1300 4 : tracing::error!("policy engine {} panicked; denying", engine.name());
1301 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic")
1302 4 : .increment(1);
1303 4 : Decision::Deny(ENGINE_FAILED.to_owned())
1304 : }
1305 : Err(_) => {
1306 12 : tracing::error!(
1307 : "policy engine {} did not answer within {:?}; denying",
1308 0 : engine.name(),
1309 0 : *TIMEOUT
1310 : );
1311 12 : metrics::counter!("antares_policy_failures_total", "reason" => "timeout")
1312 12 : .increment(1);
1313 12 : Decision::Deny(ENGINE_TIMED_OUT.to_owned())
1314 : }
1315 : }
1316 : }
1317 : // The browser build has no timer to race against and aborts on panic;
1318 : // it also loads no addon, so the engine is always the built-in one.
1319 : #[cfg(target_arch = "wasm32")]
1320 : {
1321 : resolve(op.clause, engine.decide(subject, op).await)
1322 : }
1323 3528 : }
1324 :
1325 : /// [`decide`] for a batch: one answer per Operation, failing closed the
1326 : /// same way. A panic, a timeout (one [`TIMEOUT`] for the whole batch) or an
1327 : /// answer whose length does not match the batch denies every member, and
1328 : /// each answer passes [`resolve`] for its own clause.
1329 84 : pub async fn decide_each(
1330 84 : engine: &dyn PolicyEngine,
1331 84 : subject: &Subject,
1332 84 : ops: &[Operation<'_>],
1333 84 : ) -> Vec<Decision> {
1334 84 : guarded(engine, ops, || engine.decide_each(subject, ops)).await
1335 84 : }
1336 :
1337 : /// One batch answer of `engine` to `ops` from `ask`, failing closed as
1338 : /// [`decide_each`] documents, each answer passed through [`resolve`].
1339 88 : async fn guarded<'a>(
1340 88 : engine: &dyn PolicyEngine,
1341 88 : ops: &[Operation<'_>],
1342 88 : ask: impl FnOnce() -> DecisionsFuture<'a>,
1343 88 : ) -> Vec<Decision> {
1344 88 : let all = |why: &str| vec![Decision::Deny(why.to_owned()); ops.len()];
1345 : #[cfg(not(target_arch = "wasm32"))]
1346 88 : let answer = {
1347 : use futures_util::FutureExt as _;
1348 88 : let built = std::panic::catch_unwind(std::panic::AssertUnwindSafe(ask));
1349 88 : let Ok(fut) = built else {
1350 0 : tracing::error!("policy engine {} panicked; denying", engine.name());
1351 0 : metrics::counter!("antares_policy_failures_total", "reason" => "panic").increment(1);
1352 0 : return all(ENGINE_FAILED);
1353 : };
1354 88 : let guarded = std::panic::AssertUnwindSafe(fut).catch_unwind();
1355 88 : match tokio::time::timeout(*TIMEOUT, guarded).await {
1356 88 : Ok(Ok(d)) => d,
1357 : Ok(Err(_)) => {
1358 0 : tracing::error!("policy engine {} panicked; denying", engine.name());
1359 0 : metrics::counter!("antares_policy_failures_total", "reason" => "panic")
1360 0 : .increment(1);
1361 0 : return all(ENGINE_FAILED);
1362 : }
1363 : Err(_) => {
1364 0 : tracing::error!(
1365 : "policy engine {} did not answer within {:?}; denying",
1366 0 : engine.name(),
1367 0 : *TIMEOUT
1368 : );
1369 0 : metrics::counter!("antares_policy_failures_total", "reason" => "timeout")
1370 0 : .increment(1);
1371 0 : return all(ENGINE_TIMED_OUT);
1372 : }
1373 : }
1374 : };
1375 : #[cfg(target_arch = "wasm32")]
1376 : let answer = ask().await;
1377 88 : if answer.len() != ops.len() {
1378 6 : tracing::error!(
1379 : "policy engine {} answered {} decisions for {} operations; denying",
1380 0 : engine.name(),
1381 0 : answer.len(),
1382 0 : ops.len()
1383 : );
1384 6 : metrics::counter!("antares_policy_failures_total", "reason" => "length").increment(1);
1385 6 : return all(ENGINE_FAILED);
1386 82 : }
1387 82 : answer
1388 82 : .into_iter()
1389 82 : .zip(ops)
1390 296 : .map(|(d, op)| resolve(op.clause, d))
1391 82 : .collect()
1392 88 : }
1393 :
1394 : /// Whether any request of each of `clauses` could be allowed for the caller
1395 : /// of `headers`: one batch-shaped probe a surface lists its operations by
1396 : /// (a surface's listing), never a decision on a request. Each clause is asked as an
1397 : /// Operation that names nothing, through [`PolicyEngine::offers_each`],
1398 : /// failing closed as [`decide_each`] does (a panic, a timeout or a short
1399 : /// answer hides every clause), and a narrowing on a clause that cannot be
1400 : /// narrowed hides it. With no engine attached every clause is offered.
1401 8 : pub async fn offers(
1402 8 : engine: Option<&dyn PolicyEngine>,
1403 8 : tenant: &antares_model::TenantId,
1404 8 : headers: &HeaderMap,
1405 8 : clauses: &[&'static str],
1406 8 : ) -> Vec<bool> {
1407 8 : let Some(engine) = engine else {
1408 4 : return vec![true; clauses.len()];
1409 : };
1410 4 : let ops: Vec<Operation<'_>> = clauses
1411 4 : .iter()
1412 4 : .map(|clause| Operation {
1413 12 : clause,
1414 12 : ids: &[],
1415 12 : types: &[],
1416 12 : attrs: &[],
1417 12 : q: None,
1418 12 : scope_q: None,
1419 12 : geo: None,
1420 12 : body: None,
1421 12 : stored: None,
1422 12 : })
1423 4 : .collect();
1424 4 : let subject = subject_of(tenant, headers);
1425 4 : guarded(engine, &ops, || engine.offers_each(&subject, &ops))
1426 4 : .await
1427 4 : .into_iter()
1428 12 : .map(|d| !matches!(d, Decision::Deny(_)))
1429 4 : .collect()
1430 8 : }
1431 :
1432 : /// Ask the engine about one notification, and fail closed: an engine that
1433 : /// panics drops the notification. The document it was handed may already be
1434 : /// half-edited, and the broker cannot know which half — so it is not sent.
1435 52 : pub fn pre_notify(
1436 52 : engine: &dyn PolicyEngine,
1437 52 : subject: &Subject,
1438 52 : sub: &Value,
1439 52 : notification: &mut Value,
1440 52 : ) -> NotifyDecision {
1441 52 : match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1442 52 : engine.pre_notify(subject, sub, notification)
1443 52 : })) {
1444 48 : Ok(d) => d,
1445 : Err(_) => {
1446 4 : tracing::error!(
1447 : "policy engine {} panicked on a notification; dropping it",
1448 0 : engine.name()
1449 : );
1450 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic").increment(1);
1451 4 : NotifyDecision::Drop
1452 : }
1453 : }
1454 52 : }
1455 :
1456 : /// The ProblemDetails `type` and `title` of a refusal. Table 6.3.2-1 names
1457 : /// no access-denied error, so none is invented under the ETSI namespace: a
1458 : /// refusal is answered with this broker's own URN, documented in the book
1459 : /// and noted in the 6.3.2 ledger entry. A client that reads error types can
1460 : /// tell a policy refusal from a spec error by the namespace alone.
1461 : pub const ACCESS_DENIED_TYPE: &str = "urn:antares:error:AccessDenied";
1462 : pub const ACCESS_DENIED_TITLE: &str = "AccessDenied";
1463 :
1464 : /// The request headers copied into a [`Subject`], comma-separated
1465 : /// (`ANTARES_POLICY_SUBJECT_HEADERS`), matched case-insensitively and read
1466 : /// once at first use. Empty by default: an engine that wants an identity
1467 : /// names the header that carries it, and a broker that copied every header
1468 : /// into the subject would be handing an addon the whole request.
1469 56 : pub static SUBJECT_HEADERS: LazyLock<Vec<String>> = LazyLock::new(|| {
1470 56 : std::env::var("ANTARES_POLICY_SUBJECT_HEADERS")
1471 56 : .unwrap_or_default()
1472 56 : .split(',')
1473 56 : .map(|s| s.trim().to_ascii_lowercase())
1474 56 : .filter(|s| !s.is_empty())
1475 56 : .collect()
1476 56 : });
1477 :
1478 : /// Who this request is from, as far as the seam is concerned: the tenant it
1479 : /// addresses and the headers a deployment named. A header the request does
1480 : /// not carry is simply absent — the seam invents nothing.
1481 9204 : pub fn subject_of(tenant: &antares_model::TenantId, headers: &HeaderMap) -> Subject {
1482 9204 : let mut carried = Vec::new();
1483 9204 : for name in SUBJECT_HEADERS.iter() {
1484 394 : for value in headers.get_all(name.as_str()) {
1485 280 : if let Ok(v) = value.to_str() {
1486 280 : carried.push((name.clone(), v.to_owned()));
1487 280 : }
1488 : }
1489 : }
1490 9204 : if let Ok(view) = VIEW.try_with(std::sync::Arc::clone) {
1491 8 : carried.push((VIEW_PAIR.to_owned(), view.to_string()));
1492 9196 : }
1493 9204 : Subject {
1494 9204 : tenant: tenant.clone(),
1495 9204 : headers: carried,
1496 9204 : }
1497 9204 : }
1498 :
1499 : /// The member a Subscription carries its creator's subject in. A
1500 : /// broker-internal member like `__context` and `__via`: a client can
1501 : /// neither set it nor read it back, it is stripped from every served
1502 : /// representation and from the 5.8.1.4 copy forwarded to a Context Source,
1503 : /// and 5.2.12 defines no such member — the whole `__` prefix is the
1504 : /// broker's, so a member added later inherits every one of those rules
1505 : /// instead of a new list to forget.
1506 : pub(crate) const SUBJECT_MEMBER: &str = "__subject";
1507 :
1508 : /// What a write by id hands the engine about the Entity it lands on.
1509 : #[derive(Debug, Default)]
1510 : pub(crate) struct PreImage {
1511 : /// The stored Entity's types joined with the ones the request body
1512 : /// names: an engine judging by type judges the Entity the write changes
1513 : /// and what it changes it to (ADR-0020).
1514 : pub types: Vec<String>,
1515 : /// The stored Entity itself ([`Operation::stored`]).
1516 : pub stored: Option<StoredImage>,
1517 : }
1518 :
1519 : /// The [`PreImage`] of a write by id. The store is read once per request,
1520 : /// and only with an engine attached, so vanilla reads nothing more. An
1521 : /// absent Entity or a store error gives no stored image, and the write
1522 : /// answers it as before.
1523 : ///
1524 : /// Decided (worker-2, owner may override): the write still reads
1525 : /// the Entity again inside its own store operation. Handing it this image,
1526 : /// read outside the store's transaction, would turn a concurrent write
1527 : /// into a lost update; the gate's read is the one the type lookup already
1528 : /// made, so no read is added.
1529 17744 : pub(crate) async fn pre_image(
1530 17744 : st: &crate::state::AppState,
1531 17744 : tenant: &antares_model::TenantId,
1532 17744 : id: &str,
1533 17744 : body: Option<&Value>,
1534 17744 : ) -> Result<PreImage, antares_model::NgsiError> {
1535 17744 : if st.policy.is_none() {
1536 17460 : return Ok(PreImage::default());
1537 284 : }
1538 : // A failed read is the request's failure, never an absent Entity: a
1539 : // missing pre-image would let a rule keyed on the stored types or
1540 : // headers decide over nothing (fail-closed, ADR-0020).
1541 284 : let stored = st
1542 284 : .store
1543 284 : .get(tenant, antares_store::Kind::Entity, id)
1544 284 : .await?;
1545 282 : Ok(image(stored, body))
1546 17744 : }
1547 :
1548 : /// The [`PreImage`] of a temporal write by id (5.6.12 to 5.6.16):
1549 : /// the Entity's temporal evolution, which a history-only Entity has
1550 : /// without a current state, else its current state. Read only with an
1551 : /// engine attached, as [`pre_image`].
1552 3760 : pub(crate) async fn temporal_pre_image(
1553 3760 : st: &crate::state::AppState,
1554 3760 : tenant: &antares_model::TenantId,
1555 3760 : id: &str,
1556 3760 : ) -> Result<PreImage, antares_model::NgsiError> {
1557 3760 : if st.policy.is_none() {
1558 3664 : return Ok(PreImage::default());
1559 96 : }
1560 96 : match st.temporal.get(tenant, id).await? {
1561 56 : Some(evolution) => Ok(image(Some(evolution), None)),
1562 38 : None => pre_image(st, tenant, id, None).await,
1563 : }
1564 3760 : }
1565 :
1566 : /// The image of a stored document, with the types the body adds.
1567 606 : pub(crate) fn image(stored: Option<Value>, body: Option<&Value>) -> PreImage {
1568 606 : let mut types: Vec<String> = Vec::new();
1569 648 : for doc in stored.as_ref().into_iter().chain(body) {
1570 650 : for t in entity_types(doc) {
1571 554 : if !types.iter().any(|o| o == t) {
1572 498 : types.push(t.to_owned());
1573 498 : }
1574 : }
1575 : }
1576 606 : PreImage {
1577 606 : types,
1578 606 : stored: stored.as_ref().map(StoredImage::of),
1579 606 : }
1580 606 : }
1581 :
1582 : /// The stored form of a subject: only the headers a deployment named, since
1583 : /// the tenant is where the subscription already lives. `None` when the
1584 : /// subject carries nothing, so a deployment that named no header stores no
1585 : /// member at all.
1586 5510 : pub(crate) fn subject_member(subject: &Subject) -> Option<Value> {
1587 5510 : (!subject.headers.is_empty())
1588 5510 : .then(|| serde_json::to_value(&subject.headers).ok())
1589 5510 : .flatten()
1590 5510 : }
1591 :
1592 : /// The subject a notification is delivered under: the one stored when the
1593 : /// subscription was created. A subscription created before the deployment
1594 : /// named its headers — or by the broker itself, for the 5.8.1.4 internal
1595 : /// copies — simply has none, and the engine is asked about a subject with
1596 : /// no headers rather than about the wrong one.
1597 278 : pub(crate) fn stored_subject(tenant: &antares_model::TenantId, sub: &Value) -> Subject {
1598 : Subject {
1599 278 : tenant: tenant.clone(),
1600 278 : headers: sub
1601 278 : .get(SUBJECT_MEMBER)
1602 278 : .cloned()
1603 278 : .and_then(|v| serde_json::from_value(v).ok())
1604 278 : .unwrap_or_default(),
1605 : }
1606 278 : }
1607 :
1608 : /// Remove every broker-internal member from a document about to be served.
1609 : ///
1610 : /// The `__` prefix is the broker's: the notification `@context` (5.8.6), the
1611 : /// 6.3.18 Via chain, a snapshot's synthetic tenant and [`SUBJECT_MEMBER`]
1612 : /// all live under it, and no NGSI-LD data type defines a member there. One
1613 : /// predicate rather than a list per document type, so a member added later
1614 : /// is hidden by construction instead of by remembering every serve point.
1615 508 : pub(crate) fn strip_internal(doc: &mut Value) {
1616 508 : if let Some(o) = doc.as_object_mut() {
1617 4403 : o.retain(|k, _| !k.starts_with("__"));
1618 0 : }
1619 508 : }
1620 :
1621 : /// Whether a stored document was made by this subject. Compared on the
1622 : /// headers alone: the tenant is where the document already lives, and a
1623 : /// deployment that named no header has no subjects to tell apart — every
1624 : /// request is the same one, and everything it stored is its own.
1625 238 : pub(crate) fn belongs_to(doc: &Value, subject: &Subject) -> bool {
1626 238 : stored_subject(&subject.tenant, doc).headers == subject.headers
1627 238 : }
1628 :
1629 : /// A refusal on its way out of a handler. Its own type rather than an
1630 : /// `ApiError`, so this module names nothing above it and stays a leaf; the
1631 : /// `?` in a handler converts it through `From`.
1632 : #[derive(Debug, Clone, PartialEq)]
1633 : pub struct Denied(pub String);
1634 :
1635 : /// The one call a handler makes. Every operation passes through it exactly
1636 : /// once, which is what `every_route_asks_the_policy_engine_exactly_once` asserts by
1637 : /// walking the router with a counting engine.
1638 : ///
1639 : /// The returned [`Filter`] is what the answer has to be narrowed by; an
1640 : /// allow returns the empty filter, which narrows nothing.
1641 3376 : pub async fn gate(
1642 3376 : engine: &dyn PolicyEngine,
1643 3376 : tenant: &antares_model::TenantId,
1644 3376 : headers: &HeaderMap,
1645 3376 : op: &Operation<'_>,
1646 3376 : ) -> Result<Filter, Denied> {
1647 3376 : match decide(engine, &subject_of(tenant, headers), op).await {
1648 2346 : Decision::Allow => Ok(Filter::default()),
1649 : // On a write by id the filter is the condition ([`Condition::of`]);
1650 : // `resolve` lets an `AllowIf` through nowhere else, and turns a
1651 : // `Filter` there into a deny, so a handler of such a clause reads
1652 : // what it gets back as the condition and nothing else.
1653 910 : Decision::Filter(f) | Decision::AllowIf(f) => Ok(f),
1654 120 : Decision::Deny(why) => Err(Denied(why)),
1655 : }
1656 3376 : }
1657 :
1658 : /// [`gate`](fn@gate) for a batch: per member the refusal reason, or the
1659 : /// condition its stored Entity has to meet (empty when there is none). A
1660 : /// batch writes its members whole, so a `Filter` on a member is a refusal
1661 : /// ([`resolve`]: no batch clause is FILTERABLE).
1662 76 : pub async fn gate_each(
1663 76 : engine: &dyn PolicyEngine,
1664 76 : tenant: &antares_model::TenantId,
1665 76 : headers: &HeaderMap,
1666 76 : ops: &[Operation<'_>],
1667 76 : ) -> Vec<Result<Filter, String>> {
1668 76 : decide_each(engine, &subject_of(tenant, headers), ops)
1669 76 : .await
1670 76 : .into_iter()
1671 280 : .map(|d| match d {
1672 28 : Decision::Deny(why) => Err(why),
1673 36 : Decision::AllowIf(f) => Ok(f),
1674 216 : Decision::Allow | Decision::Filter(_) => Ok(Filter::default()),
1675 280 : })
1676 76 : .collect()
1677 76 : }
1678 :
1679 : /// A stored Subscription or Registration as a [`Condition`] reads it: `id`,
1680 : /// `type` and `scope` as an Entity carries them, every other member one
1681 : /// Property holding its value, and `subject` the carried header values of
1682 : /// the subject that created it ([`SUBJECT_MEMBER`]), one instance each, so
1683 : /// an engine selects the caller's own with `q=subject=="<its value>"`. A
1684 : /// member a client wrote under `subject` is not the subject and is left
1685 : /// out, as are the broker's other `__` members.
1686 : // Values only, without the header names; an engine whose two
1687 : // headers can carry the same value tells them apart by a rule's ids.
1688 238 : fn record_image(doc: &Value) -> Value {
1689 238 : let ctx = crate::repr::policy_context();
1690 238 : let mut out = serde_json::Map::new();
1691 1916 : for (k, v) in doc.as_object().into_iter().flatten() {
1692 1916 : match k.as_str() {
1693 1916 : "id" | "scope" => {
1694 238 : out.insert(k.clone(), v.clone());
1695 238 : }
1696 1678 : "type" => {
1697 238 : let t = v
1698 238 : .as_str()
1699 238 : .map_or_else(|| v.clone(), |t| ctx.expand_key(t).into());
1700 238 : out.insert(k.clone(), t);
1701 : }
1702 1440 : "subject" => {}
1703 1404 : k if k.starts_with("__") => {}
1704 1046 : k => {
1705 1046 : out.insert(ctx.expand_key(k), serde_json::json!([{ "value": v }]));
1706 1046 : }
1707 : }
1708 : }
1709 238 : let mut carried: Vec<(String, String)> = doc
1710 238 : .get(SUBJECT_MEMBER)
1711 238 : .cloned()
1712 238 : .and_then(|v| serde_json::from_value(v).ok())
1713 238 : .unwrap_or_default();
1714 : // The view is where the creator came from, not who they are: a view
1715 : // id equal to some user's name must not select that user's documents.
1716 238 : carried.retain(|(name, _)| name != VIEW_PAIR);
1717 238 : if !carried.is_empty() {
1718 234 : let values = carried
1719 234 : .into_iter()
1720 234 : .map(|(_, v)| serde_json::json!({ "value": v }));
1721 234 : out.insert(ctx.expand_key("subject"), values.collect());
1722 4 : }
1723 238 : Value::Object(out)
1724 238 : }
1725 :
1726 : /// What a write by id needs of the Entity it lands on
1727 : /// ([`Decision::AllowIf`]), in the form a stored document is read in. The
1728 : /// handler checks it inside the store's own read-modify-write, on the
1729 : /// document the store is about to change.
1730 : ///
1731 : /// The same selection a read's narrowing makes ([`crate::repr::PolicyReach`]),
1732 : /// judged synchronously on one document.
1733 : pub(crate) type Condition = crate::repr::PolicyReach;
1734 :
1735 : impl Condition {
1736 : /// The selection `f` makes, whatever it projects.
1737 25321 : pub(crate) fn new(f: &Filter) -> Self {
1738 25321 : Self {
1739 25321 : q: f.q.as_deref().cloned(),
1740 25321 : scope_q: f.scope_q.clone(),
1741 25321 : areas: f.areas().cloned().collect(),
1742 25321 : rules: crate::repr::type_rules(f),
1743 25321 : }
1744 25321 : }
1745 :
1746 : /// The condition `f` sets, or `None` when it selects every Entity.
1747 25321 : pub(crate) fn of(f: &Filter) -> Option<Self> {
1748 25321 : let c = Self::new(f);
1749 25321 : (!c.selects_every()).then_some(c)
1750 25321 : }
1751 :
1752 : /// Whether the condition selects every Entity: no `q`, no `scopeQ`, no
1753 : /// area and no rules.
1754 25321 : pub(crate) fn selects_every(&self) -> bool {
1755 25321 : self.q.is_none() && self.scope_q.is_none() && self.areas.is_empty() && self.rules.is_none()
1756 25321 : }
1757 :
1758 : /// Whether the stored document `doc` is selected. A linked path in a
1759 : /// `q` never holds, as in a rule: the check runs
1760 : /// inside the store's write and reads no other Entity.
1761 368 : pub(crate) fn holds(&self, doc: &Value) -> bool {
1762 368 : let ctx = crate::repr::policy_context();
1763 368 : crate::repr::shown(self.rules.as_deref(), doc).is_some()
1764 362 : && self
1765 362 : .scope_q
1766 362 : .as_deref()
1767 362 : .is_none_or(|s| crate::scope_matches(s, doc))
1768 362 : && self.areas.iter().all(|g| g.matches(doc, ctx))
1769 362 : && self.q.as_ref().is_none_or(|q| {
1770 356 : !q.has_linked_paths() && antares_ql::eval::eval_q(q, doc, ctx, &|_| None)
1771 356 : })
1772 368 : }
1773 :
1774 : /// Whether a Subscription or Registration is selected: the
1775 : /// condition read over [`record_image`] of the stored document.
1776 230 : pub(crate) fn holds_for_record(&self, doc: &Value) -> bool {
1777 230 : self.holds(&record_image(doc))
1778 230 : }
1779 :
1780 : /// Inside the store closure: an Entity the condition does not select is
1781 : /// not found, the answer an absent one gets.
1782 14242 : pub(crate) fn pre(
1783 14242 : cond: Option<&Self>,
1784 14242 : id: &str,
1785 14242 : doc: &Value,
1786 14242 : ) -> Result<(), antares_model::NgsiError> {
1787 14242 : if cond.is_none_or(|c| c.holds(doc)) {
1788 14208 : Ok(())
1789 : } else {
1790 34 : Err(antares_model::NgsiError::ResourceNotFound(format!(
1791 34 : "entity {id} not found"
1792 34 : )))
1793 : }
1794 14242 : }
1795 :
1796 : /// [`Condition::pre`] on a Temporal Evolution, inside the temporal
1797 : /// store's closure: judged on [`latest_image`], and one it does not
1798 : /// select is the absent evolution the temporal writes answer.
1799 110 : pub(crate) fn pre_evolution(
1800 110 : cond: Option<&Self>,
1801 110 : id: &str,
1802 110 : doc: &Value,
1803 110 : ) -> Result<(), antares_model::NgsiError> {
1804 110 : if cond.is_none_or(|c| c.holds(&latest_image(doc))) {
1805 94 : Ok(())
1806 : } else {
1807 16 : Err(antares_model::NgsiError::ResourceNotFound(format!(
1808 16 : "temporal entity {id} not found"
1809 16 : )))
1810 : }
1811 110 : }
1812 :
1813 : /// A conditioned write that would reach a Context Source is refused:
1814 : /// nothing in this store can judge the Source's copy.
1815 17808 : pub(crate) fn unforwarded(cond: Option<&Self>, forwards: bool) -> Result<(), Denied> {
1816 17808 : if cond.is_some() && forwards {
1817 8 : Err(Denied(CONDITION_NOT_FORWARDED.to_owned()))
1818 : } else {
1819 17800 : Ok(())
1820 : }
1821 17808 : }
1822 : }
1823 :
1824 : /// A Temporal Evolution as the Entity it now is: each Attribute keeps its
1825 : /// latest instance per `datasetId` (the greatest `observedAt`, else
1826 : /// `createdAt`; the later one on a tie) and a temporal `scope` the value of
1827 : /// its latest instance. A condition judged on every instance would let a
1828 : /// value the Entity held once select it for good.
1829 : // Instances without a time compare as the earliest; the temporal
1830 : // writes stamp `createdAt` on every instance they store.
1831 32 : pub(crate) fn latest_image(doc: &Value) -> Value {
1832 96 : let time = |i: &Value| {
1833 96 : ["observedAt", "createdAt"]
1834 96 : .iter()
1835 96 : .find_map(|k| i.get(*k).and_then(Value::as_str))
1836 96 : .map(antares_model::dt_key)
1837 96 : .unwrap_or_default()
1838 96 : };
1839 32 : let mut out = serde_json::Map::new();
1840 172 : for (k, v) in doc.as_object().into_iter().flatten() {
1841 172 : let instances = match v.as_array() {
1842 84 : Some(a) if !antares_model::is_meta(k) || k == "scope" => a,
1843 : _ => {
1844 116 : out.insert(k.clone(), v.clone());
1845 116 : continue;
1846 : }
1847 : };
1848 56 : if !instances.iter().all(Value::is_object) {
1849 0 : out.insert(k.clone(), v.clone());
1850 0 : continue;
1851 56 : }
1852 56 : let mut latest: Vec<&Value> = Vec::new();
1853 108 : for i in instances {
1854 108 : let ds = i.get("datasetId");
1855 108 : match latest.iter_mut().find(|l| l.get("datasetId") == ds) {
1856 48 : Some(l) if time(i) >= time(l) => *l = i,
1857 28 : Some(_) => {}
1858 60 : None => latest.push(i),
1859 : }
1860 : }
1861 56 : let kept = if k == "scope" {
1862 4 : latest
1863 4 : .first()
1864 4 : .and_then(|i| i.get("value"))
1865 4 : .cloned()
1866 4 : .unwrap_or(Value::Null)
1867 : } else {
1868 52 : Value::Array(latest.into_iter().cloned().collect())
1869 : };
1870 56 : out.insert(k.clone(), kept);
1871 : }
1872 32 : Value::Object(out)
1873 32 : }
1874 :
1875 : /// The members of a JSON object, or nothing when it is not one.
1876 : #[cfg(any(test, feature = "test-kit"))]
1877 52 : fn members(v: &Value) -> Vec<String> {
1878 52 : v.as_object()
1879 52 : .map(|o| o.keys().cloned().collect())
1880 52 : .unwrap_or_default()
1881 52 : }
1882 :
1883 : /// The contract every engine has to hold, so an addon's own tests can call
1884 : /// it. It asserts the three things an engine can actually get wrong — it
1885 : /// stops answering, it hands back an answer the seam has to override, or it
1886 : /// puts something into a notification that was not there — and it asserts
1887 : /// them through the seam, so an engine that passes here passes as the
1888 : /// broker will call it.
1889 : ///
1890 : /// The core runs it against [`AllowAll`]; `examples/plugin-example` runs it
1891 : /// against the reference engine.
1892 : #[cfg(any(test, feature = "test-kit"))]
1893 12 : pub async fn run_policy_contract(engine: &dyn PolicyEngine) {
1894 12 : let name = engine.name();
1895 12 : assert!(!name.is_empty(), "an engine answers to a name");
1896 :
1897 12 : let subject = Subject {
1898 12 : tenant: antares_model::TenantId::default(),
1899 12 : headers: vec![("X-Subject".into(), "someone".into())],
1900 12 : };
1901 12 : let doc = serde_json::json!({
1902 12 : "id": "urn:ngsi-ld:Vehicle:1",
1903 12 : "type": "Vehicle",
1904 12 : "speed": {"type": "Property", "value": 10},
1905 12 : "brand": {"type": "Property", "value": "Skoda"}
1906 : });
1907 :
1908 : // The engine answers, and the answer is its own: a deny carrying one of
1909 : // the seam's own reasons means it timed out or panicked instead.
1910 36 : for clause in ["5.6.1", "5.7.2", "5.8.1"] {
1911 36 : match decide(engine, &subject, &Operation::new(clause)).await {
1912 24 : Decision::Allow => {}
1913 8 : Decision::Deny(why) => assert!(
1914 8 : why != ENGINE_TIMED_OUT && why != ENGINE_FAILED,
1915 : "{name}: {clause} was denied by the seam, not by the engine: {why}"
1916 : ),
1917 4 : Decision::Filter(f) => {
1918 4 : let mut narrowed = doc.clone();
1919 4 : f.project(&mut narrowed);
1920 4 : let before = members(&doc);
1921 12 : for key in members(&narrowed) {
1922 12 : assert!(
1923 12 : before.contains(&key),
1924 : "{name}: the filter for {clause} added {key:?} to the answer"
1925 : );
1926 : }
1927 : }
1928 : // `decide` answers a condition on anything but a write by id as
1929 : // a deny, so reaching here is the seam's fault, not the engine's.
1930 : // A contract harness (test-kit) fails by panicking, as asserts do.
1931 : #[allow(clippy::panic)]
1932 0 : Decision::AllowIf(_) => panic!("{name}: a condition on {clause} passed the seam"),
1933 : }
1934 : }
1935 :
1936 : // An operation the broker performs whole is allowed or refused, never
1937 : // done to less than it says: WHOLE_TENANT because there is no narrowed
1938 : // form of "delete everything", and the writes because no handler there
1939 : // reads a Filter at all (FILTERABLE).
1940 96 : for clause in WHOLE_TENANT
1941 12 : .iter()
1942 12 : .chain(["5.6.1", "5.6.6", "5.8.1", "5.9.2"].iter())
1943 : {
1944 96 : assert!(
1945 96 : !matches!(
1946 96 : decide(engine, &subject, &Operation::new(clause)).await,
1947 : Decision::Filter(_)
1948 : ),
1949 : "{name}: {clause} was answered with a filter"
1950 : );
1951 : }
1952 :
1953 : // `pre_notify` holds the notification by `&mut`, which is the one place
1954 : // an engine can widen rather than narrow: a member it puts there is a
1955 : // member no subscriber asked for and no store answered with.
1956 12 : let sub = serde_json::json!({"id": "urn:ngsi-ld:Subscription:1", "type": "Subscription"});
1957 12 : let mut notification = serde_json::json!({
1958 12 : "id": "urn:ngsi-ld:Notification:1",
1959 12 : "type": "Notification",
1960 12 : "subscriptionId": "urn:ngsi-ld:Subscription:1",
1961 12 : "notifiedAt": "2026-01-01T00:00:00Z",
1962 12 : "data": [doc.clone()]
1963 : });
1964 12 : let before = members(¬ification);
1965 12 : let entity_members = members(&doc);
1966 12 : if pre_notify(engine, &subject, &sub, &mut notification) != NotifyDecision::Drop {
1967 56 : for key in members(¬ification) {
1968 56 : assert!(
1969 56 : before.contains(&key),
1970 : "{name}: pre_notify added {key:?} to the notification"
1971 : );
1972 : }
1973 8 : let served = notification
1974 8 : .get("data")
1975 8 : .and_then(Value::as_array)
1976 8 : .cloned()
1977 8 : .unwrap_or_default();
1978 8 : for entity in &served {
1979 28 : for key in members(entity) {
1980 28 : assert!(
1981 28 : entity_members.contains(&key),
1982 : "{name}: pre_notify added {key:?} to a notified Entity"
1983 : );
1984 : }
1985 : }
1986 0 : }
1987 :
1988 : // And the seam stops waiting for this engine, whatever it does.
1989 : #[cfg(not(target_arch = "wasm32"))]
1990 : {
1991 8 : let slow = Slow(engine);
1992 8 : assert_eq!(
1993 8 : decide(&slow, &subject, &Operation::new("5.6.1")).await,
1994 8 : Decision::Deny(ENGINE_TIMED_OUT.to_owned()),
1995 : "{name}: the seam waited past its timeout instead of denying"
1996 : );
1997 : }
1998 8 : }
1999 :
2000 : /// The same engine, one timeout slower: what the contract wraps it in to
2001 : /// prove the seam stops waiting.
2002 : #[cfg(all(any(test, feature = "test-kit"), not(target_arch = "wasm32")))]
2003 : struct Slow<'e>(&'e dyn PolicyEngine);
2004 :
2005 : #[cfg(all(any(test, feature = "test-kit"), not(target_arch = "wasm32")))]
2006 : impl PolicyEngine for Slow<'_> {
2007 0 : fn name(&self) -> &str {
2008 0 : self.0.name()
2009 0 : }
2010 :
2011 8 : fn decide<'a>(&'a self, subject: &'a Subject, op: &'a Operation<'a>) -> DecisionFuture<'a> {
2012 8 : Box::pin(async move {
2013 8 : tokio::time::sleep(
2014 8 : TIMEOUT
2015 8 : .saturating_mul(2)
2016 8 : .saturating_add(Duration::from_millis(50)),
2017 8 : )
2018 8 : .await;
2019 0 : self.0.decide(subject, op).await
2020 0 : })
2021 8 : }
2022 :
2023 0 : fn pre_notify(&self, s: &Subject, sub: &Value, n: &mut Value) -> NotifyDecision {
2024 0 : self.0.pre_notify(s, sub, n)
2025 0 : }
2026 : }
2027 :
2028 : /// The most id segments a probe narrows by; past them it narrows by the
2029 : /// literal alone.
2030 : const MAX_ID_SEGMENTS: usize = 64;
2031 :
2032 : /// The exact id prefixes a rule's one id pattern forces, from the segments
2033 : /// the Tenant's ids show ([`antares_store::filter::id_pattern_segments`]):
2034 : /// the store reads them as ranges of an index in their own order, where the
2035 : /// literal of `types_attrs::id_narrowing` filtered a walk of the whole type, which for
2036 : /// a few late matches still took 10 s on a deployment. `None` when the
2037 : /// pattern does not split or the store keeps no segments.
2038 672 : pub(crate) async fn id_prefixes(
2039 672 : st: &crate::state::AppState,
2040 672 : tenant: &antares_model::TenantId,
2041 672 : rule: Option<&Rule>,
2042 672 : ) -> Result<Option<Vec<String>>, antares_model::NgsiError> {
2043 16 : let Some(seg) =
2044 672 : rule.filter(|r| r.ids.is_empty())
2045 672 : .and_then(|r| match r.id_patterns.as_slice() {
2046 28 : [p] => antares_store::filter::id_pattern_segments(p),
2047 410 : _ => None,
2048 438 : })
2049 : else {
2050 656 : return Ok(None);
2051 : };
2052 16 : Ok(st
2053 16 : .store
2054 16 : .entity_id_segments(tenant, &seg.lead, seg.sep, MAX_ID_SEGMENTS)
2055 16 : .await?
2056 16 : .map(|segs| {
2057 16 : segs.into_iter()
2058 20 : .map(|s| format!("{s}{}", seg.tail))
2059 16 : .collect()
2060 16 : }))
2061 672 : }
2062 :
2063 : /// The exact id prefixes every rule of `rules` forces, united: a necessary
2064 : /// condition of the rules' OR, as [`id_prefixes`] is of one rule. A query
2065 : /// under a dataset view whose Policy narrows by one `idPattern` walked every
2066 : /// Entity of the type for its few matches, the regex judged per row: 10 s and
2067 : /// the gateway's 503 for `limit=1` over 128 678 KPIs.
2068 : /// `None` when a rule that holds for anything does not split.
2069 256 : pub(crate) async fn rules_id_prefixes(
2070 256 : st: &crate::state::AppState,
2071 256 : tenant: &antares_model::TenantId,
2072 256 : rules: &[Rule],
2073 256 : ) -> Result<Option<Vec<String>>, antares_model::NgsiError> {
2074 256 : let mut all = Vec::new();
2075 : // a rule naming no type holds for nothing (`Rule::types`)
2076 256 : for rule in rules.iter().filter(|r| !r.types.is_empty()) {
2077 254 : match id_prefixes(st, tenant, Some(rule)).await? {
2078 12 : Some(ps) => all.extend(ps),
2079 242 : None => return Ok(None),
2080 : }
2081 : }
2082 14 : Ok(Some(all).filter(|a| !a.is_empty()))
2083 256 : }
2084 :
2085 : /// The `idPattern`s every rule of `rules` that holds for anything narrows
2086 : /// by, when that is all they narrow by beside their types and they all name
2087 : /// the same ones, each of which the store applies exactly
2088 : /// ([`antares_store::filter::id_pattern_sql`]): then the rules hold for
2089 : /// exactly the Entities of their types whose id matches one, and the store
2090 : /// decides, pages and counts them. A view's Policy often narrows its type
2091 : /// by one such pattern; judged per row, a count over 164 000 Trees
2092 : /// took past the gateway's 10 s.
2093 354 : pub(crate) fn shared_id_patterns(rules: &[Rule]) -> Option<&[String]> {
2094 448 : let mut typed = rules.iter().filter(|r| !r.types.is_empty());
2095 354 : let first = typed.next()?;
2096 352 : let narrows =
2097 440 : |r: &Rule| !r.is_conditional() && r.ids.is_empty() && r.id_patterns == first.id_patterns;
2098 352 : let exact = |p: &String| {
2099 22 : antares_store::filter::id_pattern_sql(p).is_some()
2100 18 : && antares_ql::regex::compile(&format!("^(?:{p})$")).is_ok()
2101 22 : };
2102 352 : (narrows(first)
2103 134 : && typed.all(narrows)
2104 114 : && !first.id_patterns.is_empty()
2105 22 : && first.id_patterns.iter().all(exact))
2106 352 : .then_some(first.id_patterns.as_slice())
2107 354 : }
2108 :
2109 : #[cfg(test)]
2110 : mod tests {
2111 : use super::*;
2112 :
2113 : /// `admits` is `shown(..).is_some()` without the set.
2114 : #[test]
2115 4 : fn admits_is_whether_shown_answers() {
2116 4 : let ctx = antares_jsonld::core_context();
2117 4 : let car = serde_json::json!({"id": "urn:ngsi-ld:Car:1", "type": "Car"});
2118 4 : let rule = |t: &str| Rule {
2119 16 : types: vec![t.to_owned()],
2120 16 : ..Rule::default()
2121 16 : };
2122 4 : let cases = [
2123 4 : Filter::default(),
2124 4 : Filter {
2125 4 : omit_by_type: BTreeMap::from([("Car".to_owned(), vec!["speed".to_owned()])]),
2126 4 : ..Filter::default()
2127 4 : },
2128 4 : Filter {
2129 4 : rules: vec![rule("Car")],
2130 4 : ..Filter::default()
2131 4 : },
2132 4 : Filter {
2133 4 : rules: vec![rule("User")],
2134 4 : ..Filter::default()
2135 4 : },
2136 4 : Filter {
2137 4 : rules: vec![rule("User"), rule("Car")],
2138 4 : ..Filter::default()
2139 4 : },
2140 4 : ];
2141 20 : let admitted: Vec<bool> = cases.iter().map(|f| f.admits(&car, &ctx)).collect();
2142 4 : assert_eq!(admitted, [true, true, true, false, true]);
2143 20 : for f in &cases {
2144 20 : assert_eq!(f.admits(&car, &ctx), f.shown(&car, &ctx).is_some(), "{f:?}");
2145 : }
2146 4 : }
2147 :
2148 : /// an engine that keeps the trait's defaults reads no rule
2149 : /// documents (ADR-0020 amendment): it says so, so `/q/policy` answers
2150 : /// 404, and a document handed to it anyway is refused, never taken.
2151 : #[test]
2152 4 : fn an_engine_on_the_defaults_reads_no_rule_documents() {
2153 4 : assert!(!AllowAll.loads_documents());
2154 4 : let refused = AllowAll.load("city", &serde_json::json!({"rules": []}));
2155 4 : assert!(
2156 4 : refused
2157 4 : .as_ref()
2158 4 : .is_err_and(|why| why.contains(BUILT_IN_NAME)),
2159 : "{refused:?}"
2160 : );
2161 4 : }
2162 :
2163 : /// `offers` answers one verdict per clause, in the clauses'
2164 : /// order, false for a Deny; with no engine every clause is offered.
2165 : #[tokio::test]
2166 4 : async fn offers_answers_each_clause_in_order() {
2167 : struct DenyRetrieve;
2168 : impl PolicyEngine for DenyRetrieve {
2169 0 : fn name(&self) -> &str {
2170 0 : "deny-retrieve"
2171 0 : }
2172 12 : fn decide<'a>(&'a self, _s: &'a Subject, op: &'a Operation<'a>) -> DecisionFuture<'a> {
2173 12 : Box::pin(std::future::ready(if op.clause == "retrieveEntity" {
2174 4 : Decision::Deny("no".into())
2175 : } else {
2176 8 : Decision::Allow
2177 : }))
2178 12 : }
2179 0 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
2180 0 : NotifyDecision::Deliver
2181 0 : }
2182 : }
2183 4 : let (tenant, headers) = (antares_model::TenantId::default(), HeaderMap::new());
2184 4 : let clauses = ["queryEntity", "retrieveEntity", "queryTemporal"];
2185 4 : assert_eq!(
2186 4 : offers(Some(&DenyRetrieve), &tenant, &headers, &clauses).await,
2187 : [true, false, true]
2188 : );
2189 4 : assert_eq!(offers(None, &tenant, &headers, &clauses).await, [true; 3]);
2190 4 : }
2191 :
2192 : /// only the caller's own q is rewritten; the engine's
2193 : /// condition may read what it hides from the caller.
2194 : #[test]
2195 4 : fn a_hidden_term_is_rewritten_and_the_engines_own_q_is_not() {
2196 4 : let filter = Filter {
2197 4 : omit: vec!["salary".into()],
2198 4 : q: Some(Box::new(antares_ql::parse_q("salary>0").expect("q"))),
2199 4 : ..Filter::default()
2200 4 : };
2201 4 : let ctx = antares_jsonld::core_context();
2202 4 : let params = std::collections::HashMap::from([(
2203 4 : "q".to_owned(),
2204 4 : "salary==1|!salary|name==\"x\"".to_owned(),
2205 4 : )]);
2206 4 : let out = filter.narrow_params(¶ms, &ctx).expect("narrowed");
2207 4 : assert_eq!(
2208 4 : out["q"],
2209 : "(salary;!salary|(salary|!salary)|name==\"x\");salary>0"
2210 : );
2211 4 : let nothing_hidden = Filter::default();
2212 4 : let out = nothing_hidden
2213 4 : .narrow_params(¶ms, &ctx)
2214 4 : .expect("narrowed");
2215 4 : assert_eq!(out["q"], params["q"]);
2216 4 : }
2217 : use futures_util::FutureExt as _;
2218 :
2219 : const NS: &str = "https://uri.etsi.org/ngsi-ld/default-context/";
2220 :
2221 712 : fn iri(n: &str) -> String {
2222 712 : format!("{NS}{n}")
2223 712 : }
2224 :
2225 : /// Zone A: the square (0,0)-(1,1).
2226 44 : fn zone_a() -> Box<GeoQuery> {
2227 44 : let params = std::collections::HashMap::from([
2228 44 : ("georel".to_owned(), "within".to_owned()),
2229 44 : ("geometry".to_owned(), "Polygon".to_owned()),
2230 44 : (
2231 44 : "coordinates".to_owned(),
2232 44 : "[[[0,0],[1,0],[1,1],[0,1],[0,0]]]".to_owned(),
2233 44 : ),
2234 44 : ]);
2235 44 : Box::new(
2236 44 : GeoQuery::from_params(¶ms)
2237 44 : .expect("geoquery")
2238 44 : .expect("present"),
2239 : )
2240 44 : }
2241 :
2242 160 : fn rule(t: &str, attrs: &[&str]) -> Rule {
2243 : Rule {
2244 160 : types: vec![iri(t)],
2245 160 : attributes: attrs.iter().map(|a| iri(a)).collect(),
2246 160 : ..Rule::default()
2247 : }
2248 160 : }
2249 :
2250 : /// `{Car in zone A: speed, location, status}`, `{Car: status}`,
2251 : /// `{User: name}`, in the internal form `repr::type_rules` produces.
2252 20 : fn zones() -> Filter {
2253 20 : Filter {
2254 20 : rules: vec![
2255 20 : Rule {
2256 20 : geo_q: Some(zone_a()),
2257 20 : attributes: vec![
2258 20 : iri("speed"),
2259 20 : antares_ql::geo::LOCATION_IRI.to_owned(),
2260 20 : iri("status"),
2261 20 : ],
2262 20 : ..rule("Car", &[])
2263 20 : },
2264 20 : rule("Car", &["status"]),
2265 20 : rule("User", &["name"]),
2266 20 : ],
2267 20 : ..Filter::default()
2268 20 : }
2269 20 : }
2270 :
2271 60 : fn car(lon: f64) -> Value {
2272 60 : json!({"id": "urn:ngsi-ld:Car:1", "type": [iri("Car")],
2273 60 : (iri("speed")): [{"type": "Property", "value": 80}],
2274 60 : (iri("status")): [{"type": "Property", "value": "ok"}],
2275 60 : "https://uri.etsi.org/ngsi-ld/location": [{"type": "GeoProperty",
2276 60 : "value": {"type": "Point", "coordinates": [lon, 0.5]}}]})
2277 60 : }
2278 :
2279 28 : fn ctx() -> antares_jsonld::Context {
2280 28 : antares_jsonld::core_context()
2281 28 : }
2282 :
2283 : /// A rule that narrows ids selects by id, and a document with no
2284 : /// id only when it narrows none.
2285 : #[test]
2286 4 : fn a_rule_selects_by_type_and_id() {
2287 4 : let any = rule("Car", &["speed"]);
2288 4 : let one = Rule {
2289 4 : ids: vec!["urn:ngsi-ld:Car:1".into()],
2290 4 : ..rule("Car", &["speed"])
2291 4 : };
2292 4 : let doc = car(0.5);
2293 4 : let anonymous = json!({"type": [iri("Car")]});
2294 4 : assert!(any.selects(&doc) && one.selects(&doc));
2295 4 : assert!(any.selects(&anonymous), "no id narrowing, no id needed");
2296 4 : assert!(!one.selects(&anonymous), "an id narrowing needs an id");
2297 4 : assert!(!one.selects(&json!({"id": "urn:ngsi-ld:Car:2", "type": [iri("Car")]})));
2298 4 : assert!(!one.selects(&json!({"id": "urn:ngsi-ld:Car:1", "type": [iri("Bus")]})));
2299 4 : }
2300 :
2301 : /// the union over the rules that hold, nothing for an
2302 : /// Entity none holds for, and an omit on any type wins.
2303 : #[test]
2304 4 : fn shown_is_the_union_of_the_rules_that_hold_and_any_omit_wins() {
2305 4 : let f = zones();
2306 4 : let ctx = ctx();
2307 4 : let inside = f.shown(&car(0.5), &ctx).expect("in view");
2308 4 : let outside = f.shown(&car(5.0), &ctx).expect("in view");
2309 4 : assert!(inside.allows(&iri("speed")) && inside.allows(&iri("status")));
2310 4 : assert!(!outside.allows(&iri("speed")), "speed only in zone A");
2311 4 : assert!(outside.allows(&iri("status")));
2312 4 : let robot = json!({"id": "urn:ngsi-ld:Robot:1", "type": [iri("Robot")]});
2313 4 : assert_eq!(f.shown(&robot, &ctx), None, "no rule holds: out");
2314 4 : let both = json!({"id": "urn:x", "type": [iri("Car"), iri("User")],
2315 4 : (iri("name")): [{"type": "Property", "value": "n"}]});
2316 4 : let shown = f.shown(&both, &ctx).expect("in view");
2317 4 : assert!(
2318 4 : shown.allows(&iri("name")) && shown.allows(&iri("status")),
2319 : "union"
2320 : );
2321 4 : let omitting = Filter {
2322 4 : omit_by_type: BTreeMap::from([(iri("User"), vec![iri("status")])]),
2323 4 : ..zones()
2324 4 : };
2325 4 : let shown = omitting.shown(&both, &ctx).expect("in view");
2326 4 : assert!(!shown.allows(&iri("status")), "an omit on any type wins");
2327 4 : let only_omit = Filter {
2328 4 : omit_by_type: omitting.omit_by_type.clone(),
2329 4 : ..Filter::default()
2330 4 : };
2331 4 : let shown = only_omit.shown(&robot, &ctx).expect("no rules: in");
2332 4 : assert!(shown.allows(&iri("serial")), "no rules: omit alone");
2333 4 : }
2334 :
2335 : /// A rule whose `q` follows a link never holds (fail closed), and a
2336 : /// rule naming no type holds for nothing.
2337 : #[test]
2338 4 : fn a_rule_reading_a_link_or_naming_no_type_never_holds() {
2339 4 : let ctx = ctx();
2340 4 : let linked = Rule {
2341 4 : q: Some(Box::new(
2342 4 : antares_ql::parse_q("ownedBy{name}==\"x\"").expect("q"),
2343 4 : )),
2344 4 : ..rule("Car", &["speed"])
2345 4 : };
2346 4 : assert!(linked.q.as_deref().is_some_and(QNode::has_linked_paths));
2347 4 : assert!(!linked.holds(&car(0.5), &ctx));
2348 4 : let negated = Rule {
2349 4 : q: Some(Box::new(antares_ql::parse_q("!ownedBy{name}").expect("q"))),
2350 4 : ..rule("Car", &["speed"])
2351 4 : };
2352 4 : assert!(!negated.holds(&car(0.5), &ctx), "not even negated");
2353 4 : assert!(!Rule::default().holds(&car(0.5), &ctx));
2354 4 : let local = Rule {
2355 4 : q: Some(Box::new(antares_ql::parse_q("speed>50").expect("q"))),
2356 4 : ..rule("Car", &["status"])
2357 4 : };
2358 4 : assert!(local.holds(&car(0.5), &ctx));
2359 4 : }
2360 :
2361 : /// the narrowing rules per rule (K22): a write names only Attributes the
2362 : /// rules that hold for its post-image show.
2363 : #[test]
2364 4 : fn write_that_adds_speed_outside_zone_a_is_refused() {
2365 4 : let f = zones();
2366 4 : let ctx = ctx();
2367 4 : let speed = iri("speed");
2368 4 : let status = iri("status");
2369 4 : assert!(f.admits_write(&car(0.5), &[&speed, &status], &ctx));
2370 4 : assert!(
2371 4 : !f.admits_write(&car(5.0), &[&speed], &ctx),
2372 : "speed outside A"
2373 : );
2374 4 : assert!(f.admits_write(&car(5.0), &[&status], &ctx));
2375 4 : let name = iri("name");
2376 4 : assert!(
2377 4 : !f.admits_write(&car(0.5), &[&speed, &name], &ctx),
2378 : "no one rule shows both"
2379 : );
2380 4 : let robot = json!({"id": "urn:ngsi-ld:Robot:1", "type": [iri("Robot")]});
2381 4 : assert!(!f.admits_write(&robot, &[], &ctx), "outside every rule");
2382 4 : }
2383 :
2384 : /// A type's list split over two rules of the same types (the Portal
2385 : /// splits past 256 names) is written whole: a write naming one name of
2386 : /// each part is admitted, and a name neither part shows is still refused
2387 : /// (17,025 open-data rows refused once).
2388 : #[test]
2389 4 : fn a_write_across_a_split_list_is_admitted_and_an_unlisted_name_refused() {
2390 4 : let ctx = ctx();
2391 4 : let f = Filter {
2392 4 : rules: vec![rule("Car", &["speed"]), rule("Car", &["status"])],
2393 4 : ..Filter::default()
2394 4 : };
2395 4 : let (speed, status, name) = (iri("speed"), iri("status"), iri("name"));
2396 4 : assert!(f.admits_write(&car(5.0), &[&speed, &status], &ctx));
2397 4 : assert!(!f.admits_write(&car(5.0), &[&speed, &name], &ctx));
2398 4 : let user = json!({"id": "urn:ngsi-ld:User:1", "type": [iri("User")]});
2399 4 : assert!(!f.admits_write(&user, &[&speed], &ctx), "no rule holds");
2400 4 : }
2401 :
2402 : /// A rule over every Attribute admits a write naming any of them, and
2403 : /// discloses nothing it reads.
2404 : #[test]
2405 4 : fn a_rule_over_every_attribute_admits_any_written_name() {
2406 4 : let ctx = ctx();
2407 4 : let whole = Rule {
2408 4 : every_attribute: true,
2409 4 : q: Some(Box::new(antares_ql::parse_q("speed>50").expect("q"))),
2410 4 : ..rule("Car", &[])
2411 4 : };
2412 4 : let f = Filter {
2413 4 : rules: vec![whole.clone()],
2414 4 : ..Filter::default()
2415 4 : };
2416 4 : let (speed, name) = (iri("speed"), iri("name"));
2417 4 : assert!(f.admits_write(&car(0.5), &[&speed, &name], &ctx));
2418 4 : let narrow = Filter {
2419 4 : rules: vec![rule("Car", &[])],
2420 4 : ..Filter::default()
2421 4 : };
2422 4 : assert!(!narrow.admits_write(&car(0.5), &[&speed], &ctx));
2423 4 : assert!(whole.discloses(&ctx).is_empty());
2424 4 : }
2425 :
2426 : /// The loader's disclosure report: zone A's rule reads `location` and
2427 : /// shows it, so it discloses nothing; a rule on `speed>100` showing only
2428 : /// `status` discloses `speed`.
2429 : #[test]
2430 4 : fn discloses_names_what_the_condition_reads_and_the_rule_does_not_show() {
2431 4 : let ctx = ctx();
2432 4 : assert!(zones().rules[0].discloses(&ctx).is_empty());
2433 4 : let fast = Rule {
2434 4 : q: Some(Box::new(antares_ql::parse_q("speed>100").expect("q"))),
2435 4 : geo_q: Some(zone_a()),
2436 4 : ..rule("Car", &["status"])
2437 4 : };
2438 4 : assert_eq!(
2439 4 : fast.discloses(&ctx),
2440 4 : vec![
2441 4 : iri("speed"),
2442 4 : "https://uri.etsi.org/ngsi-ld/location".to_owned()
2443 : ]
2444 : );
2445 4 : }
2446 :
2447 : #[test]
2448 4 : fn project_keeps_the_union_and_the_frame() {
2449 4 : let f = Filter {
2450 4 : rules: vec![
2451 4 : Rule {
2452 4 : types: vec!["Car".to_owned()],
2453 4 : attributes: vec!["speed".to_owned(), "email".to_owned()],
2454 4 : ..Rule::default()
2455 4 : },
2456 4 : Rule {
2457 4 : types: vec!["User".to_owned()],
2458 4 : attributes: vec!["name".to_owned()],
2459 4 : ..Rule::default()
2460 4 : },
2461 4 : ],
2462 4 : omit_by_type: BTreeMap::from([("User".to_owned(), vec!["email".to_owned()])]),
2463 4 : ..Filter::default()
2464 4 : };
2465 4 : let mut doc = json!({"id": "urn:x", "type": ["Car", "User"], "@context": "c",
2466 4 : "speed": 1, "name": 2, "email": 3, "serial": 4});
2467 4 : f.project(&mut doc);
2468 4 : assert_eq!(
2469 : doc,
2470 4 : json!({"id": "urn:x", "type": ["Car", "User"], "@context": "c",
2471 4 : "speed": 1, "name": 2})
2472 : );
2473 4 : }
2474 : use serde_json::json;
2475 :
2476 28 : fn subject() -> Subject {
2477 28 : Subject {
2478 28 : tenant: antares_model::TenantId::default(),
2479 28 : headers: vec![("X-Subject".into(), "a-token-shaped-string".into())],
2480 28 : }
2481 28 : }
2482 :
2483 : /// Run the contract and report whether it refused the engine.
2484 8 : async fn contract_holds(engine: &dyn PolicyEngine) -> bool {
2485 8 : let hook = std::panic::take_hook();
2486 8 : std::panic::set_hook(Box::new(|_| {}));
2487 8 : let outcome = std::panic::AssertUnwindSafe(run_policy_contract(engine))
2488 8 : .catch_unwind()
2489 8 : .await;
2490 8 : std::panic::set_hook(hook);
2491 8 : outcome.is_ok()
2492 8 : }
2493 :
2494 : #[tokio::test]
2495 4 : async fn the_built_in_engine_holds_the_contract() {
2496 4 : run_policy_contract(&AllowAll).await;
2497 4 : }
2498 :
2499 : /// An engine that narrows is what the seam is for, and the contract
2500 : /// must not stand in its way.
2501 : struct Narrowing;
2502 :
2503 : impl PolicyEngine for Narrowing {
2504 4 : fn name(&self) -> &str {
2505 4 : "narrowing"
2506 4 : }
2507 44 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
2508 44 : Box::pin(std::future::ready(Decision::Filter(Filter {
2509 44 : omit: vec!["brand".into()],
2510 44 : restricted: true,
2511 44 : ..Filter::default()
2512 44 : })))
2513 44 : }
2514 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, n: &mut Value) -> NotifyDecision {
2515 4 : if let Some(data) = n.get_mut("data").and_then(Value::as_array_mut) {
2516 4 : for entity in data {
2517 4 : if let Some(o) = entity.as_object_mut() {
2518 4 : o.remove("brand");
2519 4 : }
2520 : }
2521 0 : }
2522 4 : NotifyDecision::Filter(Filter {
2523 4 : omit: vec!["brand".into()],
2524 4 : ..Filter::default()
2525 4 : })
2526 4 : }
2527 : }
2528 :
2529 : #[tokio::test]
2530 4 : async fn an_engine_that_only_narrows_holds_the_contract() {
2531 4 : assert!(contract_holds(&Narrowing).await);
2532 4 : }
2533 :
2534 : /// The contract has teeth or it proves nothing: this engine puts a
2535 : /// member into a notification nobody asked for.
2536 : struct Widening;
2537 :
2538 : impl PolicyEngine for Widening {
2539 4 : fn name(&self) -> &str {
2540 4 : "widening"
2541 4 : }
2542 44 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
2543 44 : Box::pin(std::future::ready(Decision::Allow))
2544 44 : }
2545 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, n: &mut Value) -> NotifyDecision {
2546 4 : if let Some(o) = n.as_object_mut() {
2547 4 : o.insert("stowaway".into(), json!(true));
2548 4 : }
2549 4 : NotifyDecision::Deliver
2550 4 : }
2551 : }
2552 :
2553 : #[tokio::test]
2554 4 : async fn the_contract_refuses_an_engine_that_widens_a_notification() {
2555 4 : assert!(
2556 4 : !contract_holds(&Widening).await,
2557 4 : "the contract passed an engine that added a member to a notification"
2558 4 : );
2559 4 : }
2560 :
2561 : #[test]
2562 4 : fn a_pick_keeps_the_frame_and_drops_the_rest() {
2563 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle",
2564 4 : "speed": 1, "brand": "Skoda"});
2565 4 : Filter {
2566 4 : pick: vec!["speed".into()],
2567 4 : ..Filter::default()
2568 4 : }
2569 4 : .project(&mut doc);
2570 4 : assert_eq!(
2571 : doc,
2572 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1})
2573 : );
2574 4 : }
2575 :
2576 : #[test]
2577 4 : fn an_omit_cannot_remove_what_makes_it_an_entity() {
2578 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1});
2579 4 : Filter {
2580 4 : omit: vec!["id".into(), "type".into(), "speed".into()],
2581 4 : ..Filter::default()
2582 4 : }
2583 4 : .project(&mut doc);
2584 4 : assert_eq!(
2585 : doc,
2586 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})
2587 : );
2588 4 : }
2589 :
2590 : #[test]
2591 4 : fn a_pick_of_something_absent_adds_nothing() {
2592 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1});
2593 4 : Filter {
2594 4 : pick: vec!["mileage".into()],
2595 4 : ..Filter::default()
2596 4 : }
2597 4 : .project(&mut doc);
2598 4 : assert_eq!(
2599 : doc,
2600 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})
2601 : );
2602 4 : }
2603 :
2604 : #[test]
2605 4 : fn a_projection_leaves_a_non_object_alone() {
2606 4 : let mut doc = json!(["not", "an", "object"]);
2607 4 : Filter {
2608 4 : pick: vec!["speed".into()],
2609 4 : omit: vec!["brand".into()],
2610 4 : ..Filter::default()
2611 4 : }
2612 4 : .project(&mut doc);
2613 4 : assert_eq!(doc, json!(["not", "an", "object"]));
2614 4 : }
2615 :
2616 : #[test]
2617 4 : fn narrowing_a_whole_tenant_operation_is_a_refusal() {
2618 4 : let narrowed = Decision::Filter(Filter {
2619 4 : omit: vec!["speed".into()],
2620 4 : ..Filter::default()
2621 4 : });
2622 16 : for clause in WHOLE_TENANT {
2623 16 : assert!(matches!(
2624 16 : resolve(clause, narrowed.clone()),
2625 : Decision::Deny(_)
2626 : ));
2627 : }
2628 : // and every other clause outside FILTERABLE, because the handler
2629 : // there never reads the Filter
2630 24 : for clause in ["5.6.1", "5.6.6", "5.8.1", "5.9.2", "5.13.2", "5.8.4"] {
2631 24 : assert!(
2632 24 : matches!(resolve(clause, narrowed.clone()), Decision::Deny(_)),
2633 : "{clause} dropped a narrowing instead of refusing it"
2634 : );
2635 : }
2636 4 : }
2637 :
2638 : /// The reads that do read it keep it: the seam refuses what would be
2639 : /// dropped, and nothing more.
2640 : #[test]
2641 4 : fn a_filterable_read_keeps_its_narrowing() {
2642 4 : let narrowed = Decision::Filter(Filter {
2643 4 : omit: vec!["speed".into()],
2644 4 : ..Filter::default()
2645 4 : });
2646 4 : let selected = Decision::Filter(Filter {
2647 4 : scope_q: Some("/mine".into()),
2648 4 : ..Filter::default()
2649 4 : });
2650 60 : for clause in FILTERABLE {
2651 60 : assert_eq!(resolve(clause, selected.clone()), selected, "{clause}");
2652 60 : if RECORD_LISTS.contains(&clause) {
2653 : // a record list selects and never projects
2654 12 : assert!(matches!(
2655 12 : resolve(clause, narrowed.clone()),
2656 : Decision::Deny(_)
2657 : ));
2658 : } else {
2659 48 : assert_eq!(resolve(clause, narrowed.clone()), narrowed, "{clause}");
2660 : }
2661 : }
2662 4 : }
2663 :
2664 : #[test]
2665 4 : fn an_empty_filter_on_a_whole_tenant_operation_is_not_a_refusal() {
2666 4 : assert_eq!(
2667 4 : resolve("5.6.21", Decision::Filter(Filter::default())),
2668 : Decision::Allow
2669 : );
2670 4 : }
2671 :
2672 : struct Panicking;
2673 :
2674 : impl PolicyEngine for Panicking {
2675 0 : fn name(&self) -> &str {
2676 0 : "panicking"
2677 0 : }
2678 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
2679 4 : Box::pin(async { panic!("the engine is broken") })
2680 4 : }
2681 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
2682 4 : panic!("the engine is broken")
2683 : }
2684 : }
2685 :
2686 : /// The other half of the same failure: a synchronous engine decides in
2687 : /// the call that BUILDS the future, so a guard around the future alone
2688 : /// never sees it. The reference engine has exactly this shape.
2689 : struct PanickingEagerly;
2690 :
2691 : impl PolicyEngine for PanickingEagerly {
2692 0 : fn name(&self) -> &str {
2693 0 : "panicking-eagerly"
2694 0 : }
2695 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
2696 4 : panic!("the engine is broken")
2697 : }
2698 0 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
2699 0 : NotifyDecision::Deliver
2700 0 : }
2701 : }
2702 :
2703 : #[tokio::test]
2704 4 : async fn an_engine_that_panics_before_it_returns_a_future_denies_too() {
2705 4 : let hook = std::panic::take_hook();
2706 5 : std::panic::set_hook(Box::new(|_| {}));
2707 4 : let decided = decide(&PanickingEagerly, &subject(), &Operation::new("5.6.1")).await;
2708 4 : std::panic::set_hook(hook);
2709 4 : assert_eq!(decided, Decision::Deny(ENGINE_FAILED.to_owned()));
2710 4 : }
2711 :
2712 : #[tokio::test]
2713 4 : async fn an_engine_that_panics_denies_rather_than_allows() {
2714 4 : let hook = std::panic::take_hook();
2715 7 : std::panic::set_hook(Box::new(|_| {}));
2716 4 : let decided = decide(&Panicking, &subject(), &Operation::new("5.6.1")).await;
2717 4 : let mut notification = json!({"type": "Notification"});
2718 4 : let notified = pre_notify(&Panicking, &subject(), &json!({}), &mut notification);
2719 4 : std::panic::set_hook(hook);
2720 4 : assert_eq!(decided, Decision::Deny(ENGINE_FAILED.to_owned()));
2721 4 : assert_eq!(notified, NotifyDecision::Drop);
2722 4 : }
2723 :
2724 : /// Real time, not a paused clock: the workspace tokio is built without
2725 : /// `test-util`, so the test pays the timeout it asserts.
2726 : #[tokio::test]
2727 4 : async fn an_engine_that_never_answers_denies() {
2728 : struct Never;
2729 4 : impl PolicyEngine for Never {
2730 4 : fn name(&self) -> &str {
2731 4 : "never"
2732 4 : }
2733 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
2734 4 : Box::pin(std::future::pending())
2735 4 : }
2736 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
2737 4 : NotifyDecision::Deliver
2738 4 : }
2739 4 : }
2740 4 : assert_eq!(
2741 4 : decide(&Never, &subject(), &Operation::new("5.6.1")).await,
2742 4 : Decision::Deny(ENGINE_TIMED_OUT.to_owned())
2743 4 : );
2744 4 : }
2745 :
2746 : #[test]
2747 4 : fn the_subject_never_prints_its_header_values() {
2748 4 : let printed = format!("{:?}", subject());
2749 4 : assert!(printed.contains("X-Subject"), "{printed}");
2750 4 : assert!(
2751 4 : !printed.contains("a-token-shaped-string"),
2752 : "a header value reached a log line: {printed}"
2753 : );
2754 4 : }
2755 :
2756 : #[test]
2757 4 : fn an_operation_never_prints_the_payload() {
2758 4 : let body = json!({"id": "urn:ngsi-ld:Vehicle:1", "plate": "BB-123-XY"});
2759 4 : let printed = format!(
2760 : "{:?}",
2761 4 : Operation {
2762 4 : body: Some(&body),
2763 4 : ..Operation::new("5.6.1")
2764 4 : }
2765 : );
2766 4 : assert!(!printed.contains("BB-123-XY"), "{printed}");
2767 4 : }
2768 :
2769 : /// the policy survivors of a mutants run.
2770 : #[test]
2771 4 : fn the_timeout_takes_a_positive_integer_or_falls_back() {
2772 4 : assert_eq!(timeout_ms(None), 250);
2773 4 : assert_eq!(timeout_ms(Some("0")), 250, "zero would deny every request");
2774 4 : assert_eq!(timeout_ms(Some("x")), 250);
2775 4 : assert_eq!(timeout_ms(Some("1")), 1);
2776 4 : assert_eq!(timeout_ms(Some("300")), 300);
2777 4 : }
2778 :
2779 : #[test]
2780 4 : fn a_single_string_counts_as_a_one_element_list() {
2781 4 : let doc = json!({"type": "T", "scope": "/a"});
2782 4 : assert_eq!(entity_types(&doc), ["T"]);
2783 4 : assert_eq!(StoredImage::of(&doc).scope, ["/a"]);
2784 4 : let listed = json!({"type": ["T", "U"], "scope": ["/a", "/b"]});
2785 4 : assert_eq!(entity_types(&listed), ["T", "U"]);
2786 4 : assert_eq!(StoredImage::of(&listed).scope, ["/a", "/b"]);
2787 4 : }
2788 :
2789 : #[test]
2790 4 : fn an_operation_logs_its_shape_and_never_its_payload() {
2791 4 : let op = Operation::new("5.6.1");
2792 4 : let shown = format!("{op:?}");
2793 4 : assert!(shown.starts_with("Operation {"), "{shown}");
2794 4 : assert!(shown.contains("clause: \"5.6.1\""), "{shown}");
2795 4 : assert!(shown.contains("ids: 0"), "{shown}");
2796 4 : }
2797 :
2798 : #[test]
2799 4 : fn a_rule_is_conditional_on_any_one_of_its_conditions() {
2800 4 : let plain = rule("Car", &["speed"]);
2801 4 : assert!(!plain.is_conditional());
2802 4 : let q = Rule {
2803 4 : q: Some(Box::new(antares_ql::parse_q("speed>1").expect("q"))),
2804 4 : ..plain.clone()
2805 4 : };
2806 4 : let scope = Rule {
2807 4 : scope_q: Some("/a".to_owned()),
2808 4 : ..plain.clone()
2809 4 : };
2810 4 : assert!(q.is_conditional() && scope.is_conditional());
2811 4 : }
2812 :
2813 : #[test]
2814 4 : fn shown_nothing_admits_no_attribute() {
2815 4 : let nothing = Shown::nothing();
2816 4 : assert_eq!(nothing.only, Some(Default::default()));
2817 4 : assert!(!nothing.allows(&iri("name")));
2818 4 : }
2819 :
2820 : #[test]
2821 4 : fn a_write_is_refused_only_for_the_attribute_its_type_omits() {
2822 4 : let ctx = ctx();
2823 4 : let f = Filter {
2824 4 : omit_by_type: BTreeMap::from([(iri("User"), vec![iri("status")])]),
2825 4 : ..Filter::default()
2826 4 : };
2827 4 : let user = json!({"id": "urn:x", "type": [iri("User")]});
2828 4 : assert!(f.admits_write(&user, &[iri("name").as_str()], &ctx));
2829 4 : assert!(!f.admits_write(&user, &[iri("status").as_str()], &ctx));
2830 4 : }
2831 :
2832 : /// The image of a stored Entity names
2833 : /// its Attributes, never what frames, dates or annotates it.
2834 : #[test]
2835 4 : fn a_stored_image_names_attributes_only() {
2836 4 : let doc = json!({"id": "urn:ngsi-ld:Car:1", "type": [iri("Car")], "scope": ["/a"],
2837 4 : "createdAt": "t", "modifiedAt": "t", "deletedAt": "t",
2838 4 : "@context": [], "__tenant": "acme",
2839 4 : (iri("speed")): [{"type": "Property", "value": 1}]});
2840 4 : let image = StoredImage::of(&doc);
2841 4 : assert_eq!(image.types, vec![iri("Car")]);
2842 4 : assert_eq!(image.scope, vec!["/a".to_owned()]);
2843 4 : assert_eq!(image.attribute_names, vec![iri("speed")]);
2844 4 : }
2845 :
2846 : #[test]
2847 4 : fn mapping_the_names_maps_both_sets() {
2848 4 : let shown = Shown {
2849 4 : only: Some(std::collections::BTreeSet::from([Cow::Borrowed("a")])),
2850 4 : omitted: std::collections::BTreeSet::from([Cow::Borrowed("b")]),
2851 4 : };
2852 4 : let mapped = shown.map_names(str::to_uppercase);
2853 4 : assert_eq!(
2854 : mapped.only,
2855 4 : Some(std::collections::BTreeSet::from([Cow::Borrowed("A")]))
2856 : );
2857 4 : assert_eq!(
2858 : mapped.omitted,
2859 4 : std::collections::BTreeSet::from([Cow::Borrowed("B")])
2860 : );
2861 4 : }
2862 :
2863 : /// a notification's shown set outlives its rules unchanged; an
2864 : /// empty set would show the whole Entity.
2865 : #[test]
2866 4 : fn an_owned_shown_set_keeps_both_sets() {
2867 4 : let shown = Shown {
2868 4 : only: Some(std::collections::BTreeSet::from([Cow::Borrowed("a")])),
2869 4 : omitted: std::collections::BTreeSet::from([Cow::Borrowed("b")]),
2870 4 : };
2871 4 : let owned = shown.clone().into_owned();
2872 4 : assert_eq!(owned, shown);
2873 4 : assert!(owned.allows("a") && !owned.allows("b") && !owned.allows("c"));
2874 4 : }
2875 :
2876 : /// only a rule with no condition, ids or id patterns admits by
2877 : /// type alone, the one the store may page as a type selection.
2878 : #[test]
2879 4 : fn a_rule_admits_by_type_alone_only_without_condition_or_ids() {
2880 4 : let typed = Rule {
2881 4 : types: vec![iri("Car")],
2882 4 : ..Rule::default()
2883 4 : };
2884 4 : assert!(typed.admits_by_type_alone());
2885 12 : for narrowed in [
2886 4 : Rule {
2887 4 : ids: vec!["urn:ngsi-ld:Car:1".into()],
2888 4 : ..typed.clone()
2889 4 : },
2890 4 : Rule {
2891 4 : id_patterns: vec!["urn:.*".into()],
2892 4 : ..typed.clone()
2893 4 : },
2894 4 : Rule {
2895 4 : scope_q: Some("/a/#".into()),
2896 4 : ..typed.clone()
2897 4 : },
2898 4 : ] {
2899 12 : assert!(!narrowed.admits_by_type_alone(), "{narrowed:?}");
2900 : }
2901 4 : }
2902 :
2903 : /// the broker's `__` members of a stored Subscription or
2904 : /// Registration are no Attributes a condition reads.
2905 : #[test]
2906 4 : fn a_record_image_leaves_the_brokers_members_out() {
2907 4 : let image = record_image(&json!({"id": "urn:ngsi-ld:Subscription:1",
2908 4 : "type": "Subscription", "__tenant": "acme", "description": "d"}));
2909 4 : let keys: Vec<&String> = image
2910 4 : .as_object()
2911 4 : .into_iter()
2912 4 : .flatten()
2913 4 : .map(|(k, _)| k)
2914 4 : .collect();
2915 4 : assert_eq!(keys.len(), 3, "{image}");
2916 12 : assert!(keys.iter().all(|k| !k.contains("__")), "{image}");
2917 4 : }
2918 :
2919 : #[test]
2920 4 : fn a_filter_says_its_areas_its_rules_and_its_restriction() {
2921 4 : let both = Filter {
2922 4 : geo_q: Some(zone_a()),
2923 4 : geo_also: vec![*zone_a()],
2924 4 : ..Filter::default()
2925 4 : };
2926 4 : assert_eq!(both.areas().count(), 2);
2927 4 : let also = Filter {
2928 4 : geo_also: vec![*zone_a()],
2929 4 : ..Filter::default()
2930 4 : };
2931 4 : assert_eq!(also.areas().count(), 1);
2932 4 : assert!(!also.is_empty());
2933 :
2934 4 : assert!(!Filter::default().has_rules());
2935 4 : assert!(zones().has_rules(), "rules alone");
2936 4 : let omits = Filter {
2937 4 : omit_by_type: BTreeMap::from([(iri("User"), vec![iri("status")])]),
2938 4 : ..Filter::default()
2939 4 : };
2940 4 : assert!(omits.has_rules(), "per-type omits alone");
2941 :
2942 4 : let mut headers = HeaderMap::new();
2943 4 : Filter::default().mark("me", &mut headers);
2944 4 : assert!(headers.get(RESTRICTED_HEADER).is_none());
2945 4 : assert!(headers.get("NGSILD-Warning").is_none());
2946 4 : Filter {
2947 4 : restricted: true,
2948 4 : ..Filter::default()
2949 4 : }
2950 4 : .mark("me", &mut headers);
2951 4 : assert_eq!(
2952 4 : headers.get(RESTRICTED_HEADER).and_then(|v| v.to_str().ok()),
2953 : Some("true")
2954 : );
2955 4 : assert!(headers.get("NGSILD-Warning").is_none(), "only when set");
2956 4 : }
2957 :
2958 : /// The request's window clamped to the narrowing's, written back
2959 : /// into the parameters; nothing changes without windows, a missing or
2960 : /// invalid window stays refused, and no overlap is a window of nothing.
2961 : #[test]
2962 4 : fn a_temporal_window_is_clamped_into_the_parameters() {
2963 44 : let p = |pairs: &[(&str, &str)]| -> std::collections::HashMap<String, String> {
2964 44 : pairs
2965 44 : .iter()
2966 44 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
2967 44 : .collect()
2968 44 : };
2969 12 : let t = |s: &str| TimeBound::At(s.parse().unwrap());
2970 4 : let between = Filter {
2971 4 : temporal: Box::new([Window {
2972 4 : from: Some(t("2026-01-01T01:00:00Z")),
2973 4 : to: Some(t("2026-01-01T02:00:00Z")),
2974 4 : }]),
2975 4 : ..Filter::default()
2976 4 : };
2977 4 : let odd = p(&[("timerel", "sideways")]);
2978 4 : assert_eq!(
2979 4 : Filter::default().narrow_window(&odd, true).unwrap(),
2980 : odd,
2981 : "no windows, no reading"
2982 : );
2983 4 : assert!(
2984 4 : between.narrow_window(&p(&[]), true).is_err(),
2985 : "5.7.4 still demands a window"
2986 : );
2987 4 : assert!(between.narrow_window(&odd, false).is_err());
2988 :
2989 4 : let out = between.narrow_window(&p(&[]), false).unwrap();
2990 4 : assert_eq!(out["timerel"], "between");
2991 4 : assert_eq!(out["timeAt"], "2026-01-01T01:00:00.000Z");
2992 4 : assert_eq!(out["endTimeAt"], "2026-01-01T02:00:00.000Z");
2993 :
2994 4 : let out = between
2995 4 : .narrow_window(
2996 4 : &p(&[("timerel", "after"), ("timeAt", "2026-01-01T01:30:00Z")]),
2997 : true,
2998 : )
2999 4 : .unwrap();
3000 4 : assert_eq!(
3001 4 : (out["timerel"].as_str(), out["timeAt"].as_str()),
3002 : ("between", "2026-01-01T01:30:00Z")
3003 : );
3004 :
3005 4 : let out = between
3006 4 : .narrow_window(
3007 4 : &p(&[("timerel", "before"), ("timeAt", "2025-01-01T00:00:00Z")]),
3008 : true,
3009 : )
3010 4 : .unwrap();
3011 4 : assert_eq!(
3012 4 : out["timeAt"], out["endTimeAt"],
3013 : "no overlap: between an instant and itself"
3014 : );
3015 :
3016 4 : let open = Filter {
3017 4 : temporal: Box::new([Window {
3018 4 : from: None,
3019 4 : to: Some(t("2026-01-01T02:00:00Z")),
3020 4 : }]),
3021 4 : ..Filter::default()
3022 4 : };
3023 4 : let out = open
3024 4 : .narrow_window(
3025 4 : &p(&[
3026 4 : ("timerel", "between"),
3027 4 : ("timeAt", "2025-01-01T00:00:00Z"),
3028 4 : ("endTimeAt", "2027-01-01T00:00:00Z"),
3029 4 : ]),
3030 : true,
3031 : )
3032 4 : .unwrap();
3033 4 : assert_eq!(
3034 4 : (out["timeAt"].as_str(), out["endTimeAt"].as_str()),
3035 : ("2025-01-01T00:00:00Z", "2026-01-01T02:00:00.000Z")
3036 : );
3037 4 : let out = open.narrow_window(&p(&[]), false).unwrap();
3038 4 : assert_eq!(out["timerel"], "before");
3039 4 : assert!(!out.contains_key("endTimeAt"));
3040 4 : let empty = Filter {
3041 4 : temporal: Box::new([Window::default()]),
3042 4 : ..Filter::default()
3043 4 : };
3044 4 : assert_eq!(
3045 4 : empty.narrow_window(&p(&[]), false).unwrap(),
3046 4 : p(&[]),
3047 : "an unbounded window narrows nothing"
3048 : );
3049 :
3050 : // a relative start rounds up to the millisecond, an end down
3051 4 : let slide = Filter {
3052 4 : temporal: Box::new([Window {
3053 4 : from: Some(TimeBound::Ago(chrono::Duration::nanoseconds(1))),
3054 4 : to: Some(TimeBound::Ago(chrono::Duration::nanoseconds(1))),
3055 4 : }]),
3056 4 : ..Filter::default()
3057 4 : };
3058 4 : let out = slide.narrow_window(&p(&[]), false).unwrap();
3059 4 : assert!(antares_model::dt_key(&out["timeAt"]) >= antares_model::dt_key(&out["endTimeAt"]));
3060 :
3061 : // the windows are on observedAt: a request on another timeproperty
3062 : // is answered nothing, even one whose own window overlaps
3063 4 : let out = between
3064 4 : .narrow_window(
3065 4 : &p(&[
3066 4 : ("timerel", "after"),
3067 4 : ("timeAt", "2026-01-01T01:30:00Z"),
3068 4 : ("timeproperty", "modifiedAt"),
3069 4 : ]),
3070 : true,
3071 : )
3072 4 : .unwrap();
3073 4 : assert_eq!(out["timeAt"], out["endTimeAt"], "{out:?}");
3074 4 : assert_eq!(out["timeAt"], "2026-01-01T01:00:00.000Z", "{out:?}");
3075 4 : }
3076 :
3077 : /// A bound renders to the millisecond 4.11 takes: a start rounds up
3078 : /// (only ever by less than a millisecond), an end down, an exact
3079 : /// millisecond stays; a window is half-open on those renderings.
3080 : #[test]
3081 4 : fn a_bound_rounds_only_inward_and_a_window_is_half_open() {
3082 40 : let at = |ns: u32| {
3083 40 : TimeBound::At(chrono::DateTime::from_timestamp(1_767_225_600, ns).expect("an instant"))
3084 40 : };
3085 4 : let now = chrono::Utc::now();
3086 16 : for (ns, start, end) in [
3087 4 : (0, "00.000", "00.000"),
3088 4 : (500, "00.001", "00.000"),
3089 4 : (999_000, "00.001", "00.000"),
3090 4 : (1_000_500, "00.002", "00.001"),
3091 4 : ] {
3092 16 : assert_eq!(
3093 16 : at(ns).render(now, true),
3094 16 : format!("2026-01-01T00:00:{start}Z"),
3095 : "start at {ns} ns"
3096 : );
3097 16 : assert_eq!(
3098 16 : at(ns).render(now, false),
3099 16 : format!("2026-01-01T00:00:{end}Z"),
3100 : "end at {ns} ns"
3101 : );
3102 : }
3103 4 : let w = Window {
3104 4 : from: Some(at(0)),
3105 4 : to: Some(at(2_000_000)),
3106 4 : };
3107 4 : assert!(
3108 4 : !w.admits("2025-12-31T23:59:59.999Z", now),
3109 : "before the start"
3110 : );
3111 4 : assert!(
3112 4 : w.admits("2026-01-01T00:00:00.000Z", now),
3113 : "the start is inside"
3114 : );
3115 4 : assert!(w.admits("2026-01-01T00:00:00.001Z", now), "inside");
3116 4 : assert!(
3117 4 : !w.admits("2026-01-01T00:00:00.002Z", now),
3118 : "the end is outside"
3119 : );
3120 4 : assert!(!w.admits("2026-01-01T00:00:00.003Z", now), "after the end");
3121 4 : }
3122 :
3123 : /// an evolution is judged as the Entity it now is, per
3124 : /// Attribute and `datasetId`, whatever order its instances are stored in.
3125 : #[test]
3126 4 : fn the_latest_image_keeps_the_latest_instance_per_dataset() {
3127 20 : let i = |v: &str, at: &str, ds: Option<&str>| {
3128 20 : let mut i = serde_json::json!({"type": "Property", "value": v, "observedAt": at});
3129 20 : if let Some(ds) = ds {
3130 8 : i["datasetId"] = ds.into();
3131 12 : }
3132 20 : i
3133 20 : };
3134 4 : let doc = serde_json::json!({
3135 4 : "id": "urn:ngsi-ld:V:1",
3136 4 : "type": ["V"],
3137 4 : "zone": [
3138 4 : i("A", "2026-01-01T00:00:02Z", None),
3139 4 : i("B", "2026-01-01T00:00:01Z", None),
3140 4 : i("C", "2026-01-01T00:00:01Z", Some("urn:ds:1")),
3141 4 : i("D", "2026-01-01T00:00:03Z", Some("urn:ds:1")),
3142 4 : i("E", "2026-01-01T00:00:02Z", None),
3143 : ],
3144 4 : "scope": [
3145 4 : {"type": "Property", "value": ["/old"], "observedAt": "2026-01-01T00:00:01Z"},
3146 4 : {"type": "Property", "value": ["/new"], "observedAt": "2026-01-01T00:00:02Z"},
3147 : ],
3148 4 : "modifiedAt": "2026-01-01T00:00:03Z",
3149 : });
3150 4 : let img = latest_image(&doc);
3151 4 : let zone: Vec<&str> = img["zone"]
3152 4 : .as_array()
3153 4 : .into_iter()
3154 4 : .flatten()
3155 8 : .filter_map(|i| i["value"].as_str())
3156 4 : .collect();
3157 4 : assert_eq!(zone, ["E", "D"], "a tie goes to the later one");
3158 4 : assert_eq!(img["scope"], serde_json::json!(["/new"]));
3159 4 : assert_eq!(img["type"], doc["type"]);
3160 4 : assert_eq!(img["modifiedAt"], doc["modifiedAt"]);
3161 4 : let plain = serde_json::json!({"id": "urn:ngsi-ld:V:1", "scope": "/a"});
3162 4 : assert_eq!(latest_image(&plain), plain);
3163 4 : }
3164 :
3165 : /// The engine's warning travels as one `NGSILD-Warning` value per
3166 : /// text, and a text that could not be one header line is refused.
3167 : #[test]
3168 4 : fn a_condition_holds_only_on_a_write_by_id_and_only_as_a_selection() {
3169 4 : let zone = Filter {
3170 4 : q: Some(Box::new(antares_ql::parse_q("zone==\"A\"").expect("q"))),
3171 4 : ..Filter::default()
3172 4 : };
3173 4 : let conditioned = Decision::AllowIf(zone.clone());
3174 64 : for clause in WRITE_BY_ID {
3175 64 : assert_eq!(
3176 64 : resolve(clause, conditioned.clone()),
3177 : conditioned,
3178 : "{clause}"
3179 : );
3180 64 : assert_eq!(
3181 64 : resolve(clause, Decision::AllowIf(Filter::default())),
3182 : Decision::Allow,
3183 : "{clause}: an empty condition selects everything"
3184 : );
3185 : }
3186 24 : for clause in ["5.6.1", "5.6.7", "5.6.16", "5.6.21", "5.7.1", "5.8.1"] {
3187 24 : assert!(
3188 24 : matches!(resolve(clause, conditioned.clone()), Decision::Deny(_)),
3189 : "{clause}"
3190 : );
3191 : }
3192 20 : for projecting in [
3193 4 : Filter {
3194 4 : pick: vec!["speed".into()],
3195 4 : ..zone.clone()
3196 4 : },
3197 4 : Filter {
3198 4 : omit: vec!["speed".into()],
3199 4 : ..zone.clone()
3200 4 : },
3201 4 : Filter {
3202 4 : temporal: Box::new([Window::default()]),
3203 4 : ..zone.clone()
3204 4 : },
3205 4 : Filter {
3206 4 : omit_by_type: BTreeMap::from([("Vehicle".into(), vec!["speed".into()])]),
3207 4 : ..zone.clone()
3208 4 : },
3209 4 : Filter {
3210 4 : rules: vec![Rule {
3211 4 : types: vec!["Vehicle".into()],
3212 4 : temporal: vec![Window::default()],
3213 4 : ..Rule::default()
3214 4 : }],
3215 4 : ..Filter::default()
3216 4 : },
3217 4 : ] {
3218 20 : assert!(
3219 0 : matches!(
3220 20 : resolve("5.6.2", Decision::AllowIf(projecting.clone())),
3221 : Decision::Deny(_)
3222 : ),
3223 : "{projecting:?}"
3224 : );
3225 : }
3226 :
3227 4 : let dc = "https://uri.etsi.org/ngsi-ld/default-context/";
3228 28 : let doc = |z: &str| {
3229 28 : serde_json::json!({"id": "urn:ngsi-ld:Vehicle:v", "type": [format!("{dc}Vehicle")],
3230 28 : format!("{dc}zone"): [{"type": "Property", "value": z}]})
3231 28 : };
3232 4 : let c = Condition::of(&zone).expect("a condition");
3233 4 : assert!(c.holds(&doc("A")) && !c.holds(&doc("B")));
3234 4 : assert!(Condition::pre(Some(&c), "urn:ngsi-ld:Vehicle:v", &doc("A")).is_ok());
3235 4 : assert!(matches!(
3236 4 : Condition::pre(Some(&c), "urn:ngsi-ld:Vehicle:v", &doc("B")),
3237 4 : Err(antares_model::NgsiError::ResourceNotFound(m))
3238 4 : if m == "entity urn:ngsi-ld:Vehicle:v not found"
3239 : ));
3240 4 : assert!(Condition::pre(None, "urn:ngsi-ld:Vehicle:v", &doc("B")).is_ok());
3241 4 : let linked = Filter {
3242 4 : q: Some(Box::new(
3243 4 : antares_ql::parse_q("owner.zone==\"A\"").expect("q"),
3244 4 : )),
3245 4 : ..Filter::default()
3246 4 : };
3247 4 : assert!(
3248 4 : !Condition::of(&linked)
3249 4 : .expect("a condition")
3250 4 : .holds(&doc("A")),
3251 : "a linked path reads no other Entity inside the write, so it never holds"
3252 : );
3253 4 : let scoped = Filter {
3254 4 : scope_q: Some("/a".to_owned()),
3255 4 : ..Filter::default()
3256 4 : };
3257 4 : assert!(
3258 4 : Condition::of(&scoped).is_some(),
3259 : "a scopeQ alone selects: it is a condition"
3260 : );
3261 4 : let vehicles = Condition::of(&Filter {
3262 4 : rules: vec![Rule {
3263 4 : types: vec!["Vehicle".into()],
3264 4 : ..Rule::default()
3265 4 : }],
3266 4 : ..Filter::default()
3267 4 : })
3268 4 : .expect("a type rule alone selects: it is a condition");
3269 4 : let car = serde_json::json!({"id": "urn:ngsi-ld:Car:c", "type": [format!("{dc}Car")]});
3270 4 : assert!(vehicles.holds(&doc("B")) && !vehicles.holds(&car));
3271 4 : assert!(Condition::of(&Filter::default()).is_none());
3272 4 : assert!(Condition::unforwarded(Some(&c), false).is_ok());
3273 4 : assert!(Condition::unforwarded(None, true).is_ok());
3274 4 : assert_eq!(
3275 4 : Condition::unforwarded(Some(&c), true),
3276 4 : Err(Denied(CONDITION_NOT_FORWARDED.into()))
3277 : );
3278 4 : }
3279 :
3280 : #[test]
3281 4 : fn a_warning_is_one_header_line_or_refused() {
3282 4 : let left_out = Warning::new("types left out by the filter: Building").unwrap();
3283 4 : let mut headers = HeaderMap::new();
3284 4 : Filter {
3285 4 : warning: Some(left_out.clone()),
3286 4 : ..Filter::default()
3287 4 : }
3288 4 : .mark("me", &mut headers);
3289 4 : let sent: Vec<_> = headers
3290 4 : .get_all("NGSILD-Warning")
3291 4 : .iter()
3292 4 : .map(|v| v.to_str().unwrap())
3293 4 : .collect();
3294 4 : assert_eq!(sent, ["199 me \"types left out by the filter: Building\""]);
3295 :
3296 28 : for bad in [
3297 4 : "a\nb",
3298 4 : "a\rb",
3299 4 : "tab\there",
3300 4 : "say \"x\"",
3301 4 : "back\\slash",
3302 4 : "é",
3303 4 : "",
3304 4 : ] {
3305 28 : assert!(Warning::new(bad).is_err(), "{bad:?}");
3306 : }
3307 4 : assert!(Warning::new(&"x".repeat(Warning::MAX_LEN)).is_ok());
3308 4 : assert!(Warning::new(&"x".repeat(Warning::MAX_LEN + 1)).is_err());
3309 :
3310 4 : let other = Warning::new("other").unwrap();
3311 4 : let joined = Warning::join(Some(&left_out), Some(other.clone())).unwrap();
3312 4 : assert_eq!(joined.texts().len(), 2);
3313 4 : assert_eq!(
3314 4 : Warning::join(Some(&left_out), Some(left_out.clone())),
3315 4 : Some(left_out.clone())
3316 : );
3317 4 : assert_eq!(
3318 4 : Warning::join(None, Some(other.clone())),
3319 4 : Some(other.clone())
3320 : );
3321 4 : assert_eq!(Warning::join(Some(&other), None), Some(other));
3322 40 : let many = (0..10).fold(None, |acc: Option<Warning>, i| {
3323 40 : Warning::join(acc.as_ref(), Some(Warning::new(&format!("w{i}")).unwrap()))
3324 40 : });
3325 4 : assert_eq!(many.unwrap().texts().len(), Warning::MAX_TEXTS);
3326 4 : }
3327 :
3328 : #[test]
3329 4 : fn only_a_pick_or_an_omit_hides_a_term() {
3330 4 : let speed = iri("speed");
3331 4 : assert!(!Filter::default().hides("speed", &speed));
3332 4 : let pick = Filter {
3333 4 : pick: vec!["status".into()],
3334 4 : ..Filter::default()
3335 4 : };
3336 4 : assert!(pick.hides("speed", &speed));
3337 4 : assert!(!pick.hides("status", &iri("status")));
3338 4 : let omit = Filter {
3339 4 : omit: vec!["speed".into()],
3340 4 : ..Filter::default()
3341 4 : };
3342 4 : assert!(omit.hides("speed", &speed));
3343 4 : }
3344 :
3345 : /// Refuses upserts, allows the rest; `short` answers one decision
3346 : /// fewer than it was asked for.
3347 : struct Batch {
3348 : short: bool,
3349 : }
3350 :
3351 : impl PolicyEngine for Batch {
3352 0 : fn name(&self) -> &str {
3353 0 : "batch"
3354 0 : }
3355 24 : fn decide<'a>(&'a self, _s: &'a Subject, o: &'a Operation<'a>) -> DecisionFuture<'a> {
3356 24 : Box::pin(std::future::ready(if o.clause == "5.6.8" {
3357 12 : Decision::Deny("no upserts".into())
3358 : } else {
3359 12 : Decision::Allow
3360 : }))
3361 24 : }
3362 12 : fn decide_each<'a>(
3363 12 : &'a self,
3364 12 : s: &'a Subject,
3365 12 : ops: &'a [Operation<'a>],
3366 12 : ) -> DecisionsFuture<'a> {
3367 12 : Box::pin(async move {
3368 12 : let mut out = Vec::new();
3369 24 : for op in ops {
3370 24 : out.push(self.decide(s, op).await);
3371 : }
3372 12 : if self.short {
3373 4 : out.pop();
3374 8 : }
3375 12 : out
3376 12 : })
3377 12 : }
3378 0 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
3379 0 : NotifyDecision::Deliver
3380 0 : }
3381 : }
3382 :
3383 : #[tokio::test]
3384 4 : async fn a_batch_is_answered_member_by_member_and_a_short_answer_denies_all() {
3385 4 : let ops = [Operation::new("5.6.7"), Operation::new("5.6.8")];
3386 4 : let each = decide_each(&Batch { short: false }, &subject(), &ops).await;
3387 4 : assert_eq!(
3388 : each,
3389 4 : vec![Decision::Allow, Decision::Deny("no upserts".into())]
3390 : );
3391 4 : let short = decide_each(&Batch { short: true }, &subject(), &ops).await;
3392 4 : assert_eq!(short, vec![Decision::Deny(ENGINE_FAILED.to_owned()); 2]);
3393 4 : let gated = gate_each(
3394 4 : &Batch { short: false },
3395 4 : &antares_model::TenantId::default(),
3396 4 : &HeaderMap::new(),
3397 4 : &ops,
3398 4 : )
3399 4 : .await;
3400 4 : assert_eq!(
3401 4 : gated,
3402 4 : vec![Ok(Filter::default()), Err("no upserts".to_owned())]
3403 4 : );
3404 4 : }
3405 :
3406 : #[tokio::test]
3407 4 : async fn the_pre_image_joins_stored_and_written_types_once_each() {
3408 4 : let tenant = antares_model::TenantId::default();
3409 4 : let id = "urn:ngsi-ld:Car:pre";
3410 4 : let stored = json!({"id": id, "type": [iri("Car")],
3411 4 : (iri("speed")): [{"type": "Property", "value": 1}]});
3412 4 : let body = json!({"type": [iri("Car"), iri("Vehicle")]});
3413 4 : let bare = crate::state::AppState::new("pre-image".into());
3414 4 : bare.store
3415 4 : .create(&tenant, antares_store::Kind::Entity, id, stored)
3416 4 : .await
3417 4 : .expect("create");
3418 4 : let none = pre_image(&bare, &tenant, id, Some(&body))
3419 4 : .await
3420 4 : .expect("read");
3421 4 : assert!(
3422 4 : none.types.is_empty() && none.stored.is_none(),
3423 : "no engine, no read"
3424 : );
3425 4 : let st = bare.with_policy(std::sync::Arc::new(AllowAll));
3426 4 : let image = pre_image(&st, &tenant, id, Some(&body))
3427 4 : .await
3428 4 : .expect("read");
3429 4 : assert_eq!(image.types, vec![iri("Car"), iri("Vehicle")]);
3430 4 : assert_eq!(
3431 4 : image.stored.map(|s| s.attribute_names),
3432 4 : Some(vec![iri("speed")])
3433 4 : );
3434 4 : }
3435 :
3436 : /// a subject built inside a view's scope names the view, one
3437 : /// built outside names none, and no request can carry the pair.
3438 : #[tokio::test]
3439 4 : async fn a_subject_names_the_view_it_came_through_and_nothing_else_can() {
3440 4 : let tenant = antares_model::TenantId::default();
3441 4 : let plain = subject_of(&tenant, &HeaderMap::new());
3442 4 : assert_eq!(plain.view(), None);
3443 4 : let inside = through_view("cars", async { subject_of(&tenant, &HeaderMap::new()) }).await;
3444 4 : assert_eq!(inside.view(), Some("cars"));
3445 4 : let nested = through_view(
3446 4 : "cars",
3447 4 : through_view("bikes", async { subject_of(&tenant, &HeaderMap::new()) }),
3448 : )
3449 4 : .await;
3450 4 : assert_eq!(nested.view(), Some("bikes"), "the innermost view decides");
3451 4 : assert!(
3452 4 : axum::http::HeaderName::from_bytes(VIEW_PAIR.as_bytes()).is_err(),
3453 4 : "{VIEW_PAIR} must not be a header name a client could send"
3454 4 : );
3455 4 : }
3456 :
3457 : /// the view is stored with the subject (a notification is
3458 : /// decided for it), and a condition never reads it as one of the
3459 : /// creator's values.
3460 : #[test]
3461 4 : fn the_view_is_stored_with_the_subject_and_never_read_as_its_value() {
3462 4 : let tenant = antares_model::TenantId::default();
3463 4 : let subject = Subject {
3464 4 : tenant: tenant.clone(),
3465 4 : headers: vec![
3466 4 : ("x-caller-subject".into(), "ana".into()),
3467 4 : (VIEW_PAIR.into(), "ana".into()),
3468 4 : ],
3469 4 : };
3470 4 : let mut doc = json!({"id": "urn:ngsi-ld:Subscription:1", "type": "Subscription"});
3471 4 : if let (Some(o), Some(member)) = (doc.as_object_mut(), subject_member(&subject)) {
3472 4 : o.insert(SUBJECT_MEMBER.into(), member);
3473 4 : }
3474 4 : let stored = stored_subject(&tenant, &doc);
3475 4 : assert_eq!(stored.view(), Some("ana"));
3476 4 : assert!(belongs_to(&doc, &subject));
3477 4 : let through_other = Subject {
3478 4 : tenant: tenant.clone(),
3479 4 : headers: vec![
3480 4 : ("x-caller-subject".into(), "ana".into()),
3481 4 : (VIEW_PAIR.into(), "bikes".into()),
3482 4 : ],
3483 4 : };
3484 4 : assert!(
3485 4 : !belongs_to(&doc, &through_other),
3486 : "another view, another creator"
3487 : );
3488 4 : let image = record_image(&doc);
3489 4 : let values: Vec<&Value> = image
3490 4 : .as_object()
3491 4 : .into_iter()
3492 4 : .flatten()
3493 12 : .filter(|(k, _)| k.ends_with("subject"))
3494 4 : .flat_map(|(_, v)| v.as_array().into_iter().flatten())
3495 4 : .collect();
3496 4 : assert_eq!(values, [&json!({"value": "ana"})], "{image}");
3497 4 : }
3498 :
3499 : /// the store decides a view's reads only when every typed rule
3500 : /// narrows by the same translatable `idPattern` set and nothing else;
3501 : /// each guard of `shared_id_patterns` has its case.
3502 : #[test]
3503 4 : fn only_one_shared_translatable_pattern_set_is_left_to_the_store() {
3504 : const TREES: &str =
3505 : "^urn:ngsi-ld:Tree:[a-z0-9.-]+:town-centre:green-[a-z-]+-[0-9a-f-]{36}$";
3506 4 : let tree = |patterns: &[&str]| Rule {
3507 56 : id_patterns: patterns.iter().map(|p| (*p).to_owned()).collect(),
3508 56 : ..rule("Tree", &["name"])
3509 56 : };
3510 4 : let untyped = Rule {
3511 4 : types: Vec::new(),
3512 4 : q: Some(Box::new(antares_ql::parse_q("height>5").expect("q"))),
3513 4 : ..Rule::default()
3514 4 : };
3515 4 : let seeded = [tree(&[TREES]), untyped.clone()];
3516 4 : assert_eq!(
3517 4 : shared_id_patterns(&seeded),
3518 4 : Some([TREES.to_owned()].as_slice())
3519 : );
3520 4 : let refused: [(&str, Vec<Rule>); 10] = [
3521 4 : (
3522 4 : "q",
3523 4 : vec![Rule {
3524 4 : q: Some(Box::new(antares_ql::parse_q("height>5").expect("q"))),
3525 4 : ..tree(&[TREES])
3526 4 : }],
3527 4 : ),
3528 4 : (
3529 4 : "scopeQ",
3530 4 : vec![Rule {
3531 4 : scope_q: Some("/town".into()),
3532 4 : ..tree(&[TREES])
3533 4 : }],
3534 4 : ),
3535 4 : (
3536 4 : "geo_q",
3537 4 : vec![Rule {
3538 4 : geo_q: Some(zone_a()),
3539 4 : ..tree(&[TREES])
3540 4 : }],
3541 4 : ),
3542 4 : (
3543 4 : "geo_also",
3544 4 : vec![Rule {
3545 4 : geo_also: vec![*zone_a()],
3546 4 : ..tree(&[TREES])
3547 4 : }],
3548 4 : ),
3549 4 : (
3550 4 : "ids",
3551 4 : vec![Rule {
3552 4 : ids: vec!["urn:ngsi-ld:Tree:1".into()],
3553 4 : ..tree(&[TREES])
3554 4 : }],
3555 4 : ),
3556 4 : (
3557 4 : "other patterns",
3558 4 : vec![tree(&[TREES]), tree(&["^urn:ngsi-ld:Tree:other-[0-9]+$"])],
3559 4 : ),
3560 4 : ("no pattern after one", vec![tree(&[TREES]), tree(&[])]),
3561 4 : ("no pattern before one", vec![tree(&[]), tree(&[TREES])]),
3562 4 : ("no pattern", vec![tree(&[]), untyped]),
3563 4 : ("untranslatable", vec![tree(&["urn:x:.*"])]),
3564 4 : ];
3565 40 : for (case, rules) in &refused {
3566 40 : assert_eq!(shared_id_patterns(rules), None, "{case}");
3567 : }
3568 4 : }
3569 : }
|
