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::future::Future;
25 : use std::pin::Pin;
26 : use std::sync::LazyLock;
27 : use std::time::Duration;
28 :
29 : /// What [`PolicyEngine::decide`] hands back. Boxed because an engine that
30 : /// asks a policy server has to await, and the trait must stay object-safe:
31 : /// the broker holds one `Arc<dyn PolicyEngine>` chosen at startup.
32 : pub type DecisionFuture<'a> = Pin<Box<dyn Future<Output = Decision> + Send + 'a>>;
33 :
34 : /// How long an engine has to answer before the broker stops waiting and
35 : /// denies. Deployment knob (`ANTARES_POLICY_TIMEOUT_MS`), read once at
36 : /// first use. The default is short on purpose: the seam sits in front of
37 : /// every request, and an engine that cannot answer inside it is an outage
38 : /// either way — failing closed at 250 ms is the difference between a 403
39 : /// and a broker that stops accepting.
40 38 : pub static TIMEOUT: LazyLock<Duration> = LazyLock::new(|| {
41 38 : let ms = std::env::var("ANTARES_POLICY_TIMEOUT_MS")
42 38 : .ok()
43 38 : .and_then(|v| v.parse().ok())
44 38 : .filter(|n: &u64| *n > 0)
45 38 : .unwrap_or(250);
46 38 : Duration::from_millis(ms)
47 38 : });
48 :
49 : /// The reasons the seam denies on its own, rather than on an engine's word.
50 : /// Fixed strings: they reach the client in a ProblemDetails `detail`, and
51 : /// an engine's own text about why it failed is not the client's business.
52 : pub const ENGINE_TIMED_OUT: &str = "the policy engine did not answer in time";
53 : pub const ENGINE_FAILED: &str = "the policy engine failed";
54 :
55 : /// The clauses whose operations act on everything the tenant holds. There
56 : /// is no narrowed form of "delete every entity" or "snapshot the tenant",
57 : /// so a [`Decision::Filter`] on one of them is answered as a deny
58 : /// ([`resolve`]).
59 : pub const WHOLE_TENANT: [&str; 4] = ["5.6.21", "5.16.1", "5.16.2", "5.16.7"];
60 :
61 : /// The clauses whose handlers read the [`Filter`] an engine returns: the
62 : /// reads that can serve less than they were asked for. Every other clause
63 : /// takes its operation whole — a create either writes the Entity or does
64 : /// not — so a [`Decision::Filter`] there would be dropped, and the broker
65 : /// would perform in full an operation the engine believed it had narrowed.
66 : /// That is the one direction the seam may not fail in, so those clauses
67 : /// answer a narrowing as a deny ([`resolve`]).
68 : pub const FILTERABLE: [&str; 6] = ["5.7.1", "5.7.2", "5.7.3", "5.7.4", "5.14.4", "5.14.5"];
69 :
70 : /// Who is asking. The headers are the ones a deployment names for the
71 : /// seam to carry, copied verbatim and never interpreted: the broker does
72 : /// not know what a token is. They never leave this process — stripped from every
73 : /// forwarded request, absent from notifications, dead letters and logs,
74 : /// which is why [`std::fmt::Debug`] here prints names and not values.
75 : #[derive(Clone)]
76 : pub struct Subject {
77 : pub tenant: antares_model::TenantId,
78 : pub headers: Vec<(String, String)>,
79 : }
80 :
81 : impl std::fmt::Debug for Subject {
82 4 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83 4 : f.debug_struct("Subject")
84 4 : .field("tenant", &self.tenant)
85 4 : .field(
86 4 : "headers",
87 4 : &self.headers.iter().map(|(k, _)| k).collect::<Vec<_>>(),
88 : )
89 4 : .finish()
90 4 : }
91 : }
92 :
93 : /// What is being asked for, after negotiation and expansion: the request
94 : /// the broker is about to run, in the terms the operation itself uses.
95 : /// Every name is expanded, so an engine writes its rules against IRIs and
96 : /// not against whatever short name the caller's `@context` happened to use.
97 : pub struct Operation<'a> {
98 : /// The CIM 009 clause of the operation, e.g. `"5.6.1"`.
99 : pub clause: &'static str,
100 : /// The Entity identifiers the operation names, as the handler holds
101 : /// them; empty for a query that names none.
102 : pub ids: &'a [&'a str],
103 : pub types: &'a [String],
104 : pub attrs: &'a [String],
105 : pub q: Option<&'a QNode>,
106 : pub scope_q: Option<&'a str>,
107 : pub geo: Option<&'a GeoQuery>,
108 : /// The request body of a write, expanded. `None` for a read.
109 : pub body: Option<&'a Value>,
110 : }
111 :
112 : /// Shape and counts, never the payload: an `Operation` carries the
113 : /// caller's data, and a log line is not where it belongs.
114 : impl std::fmt::Debug for Operation<'_> {
115 4 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116 4 : f.debug_struct("Operation")
117 4 : .field("clause", &self.clause)
118 4 : .field("ids", &self.ids.len())
119 4 : .field("types", &self.types.len())
120 4 : .field("attrs", &self.attrs.len())
121 4 : .field("q", &self.q.is_some())
122 4 : .field("scope_q", &self.scope_q.is_some())
123 4 : .field("geo", &self.geo.is_some())
124 4 : .field("body", &self.body.is_some())
125 4 : .finish()
126 4 : }
127 : }
128 :
129 : impl Operation<'_> {
130 : /// The operation with nothing but its clause: what a handler that
131 : /// carries no ids, types or attributes passes, and the base every other
132 : /// call site fills in with struct-update syntax.
133 860 : pub const fn new(clause: &'static str) -> Operation<'static> {
134 860 : Operation {
135 860 : clause,
136 860 : ids: &[],
137 860 : types: &[],
138 860 : attrs: &[],
139 860 : q: None,
140 860 : scope_q: None,
141 860 : geo: None,
142 860 : body: None,
143 860 : }
144 860 : }
145 : }
146 :
147 : /// What an engine may answer. There is no third state: an operation is
148 : /// allowed, refused, or allowed over less than it asked for.
149 : #[derive(Debug, Clone, PartialEq)]
150 : pub enum Decision {
151 : Allow,
152 : /// Refused, with the engine's own reason. Answered 403 with a
153 : /// ProblemDetails whose type is an Antares URI — Table 6.3.2-1 names no
154 : /// access-denied error, so this is an Antares decision.
155 : Deny(String),
156 : /// Allowed over less. The caller cannot tell a hidden entity from an
157 : /// absent one.
158 : Filter(Filter),
159 : }
160 :
161 : /// How much less. Every member narrows and none widens: the `q` and
162 : /// `scopeQ` are conjoined with the caller's own, `pick` keeps a subset of
163 : /// the members, `omit` removes some.
164 : #[derive(Debug, Clone, Default, PartialEq)]
165 : pub struct Filter {
166 : /// Conjoined into the query the store runs, on the AST — never on the
167 : /// query string, where the 4.9 precedence of `;` against `|` would have
168 : /// to be re-derived by whoever writes the rule.
169 : pub q: Option<QNode>,
170 : /// Conjoined the same way with the request's own `scopeQ` (4.18).
171 : pub scope_q: Option<String>,
172 : /// Members removed from every document served.
173 : pub omit: Vec<String>,
174 : /// If non-empty, the only members served beside the document frame.
175 : pub pick: Vec<String>,
176 : /// Answer [`RESTRICTED_HEADER`]`: true`, so a client can know the
177 : /// answer was narrowed. Narrowing is otherwise silent.
178 : pub restricted: bool,
179 : }
180 :
181 : /// What identifies the document rather than describing it: 5.2.4 makes
182 : /// `id` and `type` mandatory members of an Entity, and an answer without
183 : /// them is not an Entity at all. A projection never removes these.
184 : const FRAME: [&str; 3] = ["id", "type", "@context"];
185 :
186 : impl Filter {
187 : /// Apply `pick`/`omit` to one document by member name. This is the
188 : /// reference semantics `run_policy_contract` holds an engine's answer
189 : /// against; a served document is projected through the request's own
190 : /// 5.5.2 representation instead (`repr::narrow_projection`), where the
191 : /// names are expanded against the request `@context` first.
192 26 : pub fn project(&self, doc: &mut Value) {
193 26 : let Some(obj) = doc.as_object_mut() else {
194 4 : return;
195 : };
196 22 : if !self.pick.is_empty() {
197 36 : obj.retain(|k, _| FRAME.contains(&k.as_str()) || self.pick.iter().any(|p| p == k));
198 12 : }
199 22 : for name in &self.omit {
200 20 : if FRAME.contains(&name.as_str()) {
201 8 : continue;
202 12 : }
203 12 : obj.remove(name);
204 : }
205 26 : }
206 :
207 : /// True when the filter would change nothing, which is how an engine
208 : /// that means "allow" can say so with an empty `Filter`.
209 98 : pub fn is_empty(&self) -> bool {
210 98 : self.q.is_none() && self.scope_q.is_none() && self.omit.is_empty() && self.pick.is_empty()
211 98 : }
212 :
213 : /// The request's own query parameters, narrowed by this decision.
214 : ///
215 : /// The `q` conjunction is made on the AST and rendered back once
216 : /// (`antares_ql`'s renderer parenthesises an `Or` inside an `And`, so
217 : /// 4.9's `;`-over-`|` precedence is the renderer's problem and not the
218 : /// rule writer's). Every consumer below reads the narrowed parameters:
219 : /// the store push-down, the local re-check that 5.7.2.4 runs over
220 : /// merged results, and the query the request is forwarded with.
221 : ///
222 : /// A `scopeQ` narrowing is set when the request carries none, and
223 : /// intersected with the request's own when it carries one. 4.19's `and`
224 : /// is over independent per-pattern predicates, so it distributes over
225 : /// the `,`/`|` disjunction and the intersection is itself a Scope
226 : /// Query — `antares_ql::scope::intersect_scope_q` writes it. Where the
227 : /// product cannot be written down the seam answers the only way that is
228 : /// not wider than the engine decided: it refuses.
229 : /// Say the answer was narrowed, when the engine asked for it to be
230 : /// said. Narrowing is otherwise silent: a caller cannot tell an Entity
231 : /// it may not see from one that is not there.
232 1846 : pub fn mark_restricted(&self, headers: &mut HeaderMap) {
233 1846 : if self.restricted {
234 12 : headers.insert(
235 12 : RESTRICTED_HEADER,
236 12 : axum::http::HeaderValue::from_static("true"),
237 12 : );
238 1834 : }
239 1846 : }
240 :
241 1430 : pub fn narrow_params(
242 1430 : &self,
243 1430 : params: &std::collections::HashMap<String, String>,
244 1430 : ) -> Result<std::collections::HashMap<String, String>, Denied> {
245 1430 : let mut out = params.clone();
246 1430 : if let Some(extra) = &self.q {
247 16 : let narrowed = match out.get("q").map(|q| antares_ql::parse_q(q)).transpose() {
248 2 : Ok(Some(own)) => QNode::And(vec![own, extra.clone()]),
249 14 : Ok(None) => extra.clone(),
250 : // the request's own `q` is parsed and refused long before a
251 : // filter reaches it; an unparsable one here is a caller that
252 : // narrowed something it never validated
253 0 : Err(_) => return Err(Denied(ENGINE_FAILED.to_owned())),
254 : };
255 16 : out.insert("q".to_owned(), narrowed.to_string());
256 1414 : }
257 1430 : if let Some(scope) = &self.scope_q {
258 14 : let narrowed = match out.get("scopeQ") {
259 6 : None => scope.clone(),
260 8 : Some(own) => antares_ql::scope::intersect_scope_q(own, scope)
261 8 : .ok_or_else(|| Denied(SCOPE_NOT_NARROWABLE.to_owned()))?,
262 : };
263 12 : out.insert("scopeQ".to_owned(), narrowed);
264 1416 : }
265 1428 : Ok(out)
266 1430 : }
267 : }
268 :
269 : /// The response header a `Filter { restricted: true }` adds. It is not in
270 : /// the `NGSILD-` namespace: that prefix is ETSI's, carries the headers
271 : /// clause 6.3 defines (`NGSILD-Tenant`, `NGSILD-EntityMap`,
272 : /// `NGSILD-Results-Count`, `NGSILD-Warning`), and a broker-invented header
273 : /// under it would collide with whatever a later version puts there — the
274 : /// same reason a refusal answers `urn:antares:error:AccessDenied` rather
275 : /// than an invented `uri.etsi.org` type.
276 : pub const RESTRICTED_HEADER: &str = "Antares-Results-Restricted";
277 :
278 : /// The refusal a `scopeQ` narrowing gets when its intersection with the
279 : /// request's own cannot be written as a Scope Query — one side selects
280 : /// nothing, or the distributed product is too large to express (see
281 : /// [`Filter::narrow_params`]).
282 : pub const SCOPE_NOT_NARROWABLE: &str =
283 : "the request's own scope query cannot be narrowed by the policy engine";
284 :
285 : /// What an engine may answer about one notification, for one subscription.
286 : #[derive(Debug, Clone, PartialEq)]
287 : pub enum NotifyDecision {
288 : Deliver,
289 : /// Send it, narrowed: the same `pick`/`omit` projection the query path
290 : /// applies, over the entities of `data`. A notification `Filter` that
291 : /// carries `q` or `scopeQ` is refused as a [`NotifyDecision::Drop`]:
292 : /// the entities were selected by the subscription's own conditions long
293 : /// before this point, there is nothing left to re-run the query against,
294 : /// and delivering the notification unfiltered would tell the engine a
295 : /// narrowing was applied that never was.
296 : Filter(Filter),
297 : /// Do not send. 5.8.6 counts this as no attempt at all — the
298 : /// notification was never sent, so it is neither a success nor a
299 : /// failure, and `timesSent` does not move.
300 : Drop,
301 : }
302 :
303 : /// The seam. One implementation ships with the broker; every other one is
304 : /// an addon crate a deployment builds itself.
305 : pub trait PolicyEngine: Send + Sync {
306 : /// The name a deployment selects the engine by, and the name
307 : /// `/q/health` reports.
308 : fn name(&self) -> &str;
309 :
310 : /// Fires once per request, after negotiation and expansion, before the
311 : /// operation and before any fan-out (ADR-0014 `on_request`).
312 : fn decide<'a>(&'a self, subject: &'a Subject, op: &'a Operation<'a>) -> DecisionFuture<'a>;
313 :
314 : /// Fires once per notification document per subscription, before the
315 : /// egress check and the send (ADR-0014 `pre_notify`). Synchronous: it
316 : /// sits inside the delivery the broker is about to make, and an engine
317 : /// that has to ask a server for every notification is a design the seam
318 : /// declines to make easy.
319 : fn pre_notify(
320 : &self,
321 : subject: &Subject,
322 : sub: &Value,
323 : notification: &mut Value,
324 : ) -> NotifyDecision;
325 : }
326 :
327 : /// The name of the engine the broker ships, which is also the name
328 : /// `/q/health` reports when no engine is attached at all: the two are the
329 : /// same decision — allow — and an operator reading it learns what the
330 : /// broker does, not which object made it happen.
331 : pub const BUILT_IN_NAME: &str = "allow-all";
332 :
333 : /// The engine the broker ships, and the one conformance is asserted
334 : /// against: it decides nothing, so the broker behaves exactly as it did
335 : /// before the seam existed.
336 : ///
337 : /// A deployment that attaches no engine does not get an instance of this:
338 : /// it gets no engine at all ([`AppState::policy`](crate::state::AppState)
339 : /// is `None`), and every gate answers the empty [`Filter`] without building
340 : /// a [`Subject`], boxing a future or arming the [`TIMEOUT`] timer. The two
341 : /// paths decide the same thing, which is what
342 : /// `the_built_in_answer_is_the_bypassed_answer` holds them to.
343 : pub struct AllowAll;
344 :
345 : impl PolicyEngine for AllowAll {
346 8 : fn name(&self) -> &str {
347 8 : BUILT_IN_NAME
348 8 : }
349 :
350 176 : fn decide<'a>(&'a self, _subject: &'a Subject, _op: &'a Operation<'a>) -> DecisionFuture<'a> {
351 176 : Box::pin(std::future::ready(Decision::Allow))
352 176 : }
353 :
354 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
355 4 : NotifyDecision::Deliver
356 4 : }
357 : }
358 :
359 : /// A `Filter` only means something where the broker can serve less than it
360 : /// was asked for, and [`FILTERABLE`] is that list. Everywhere else the
361 : /// narrowing would be silently dropped: "purge every entity, but only the
362 : /// ones you may see" is a different operation from the one the client
363 : /// asked for and would delete less than the 204 claims, and a create writes
364 : /// its Entity or does not. Those operations take the strict reading: narrow
365 : /// means refuse. An empty `Filter` asks for nothing and is an allow.
366 884 : pub fn resolve(clause: &str, decision: Decision) -> Decision {
367 192 : match decision {
368 192 : Decision::Filter(f) if !FILTERABLE.contains(&clause) => {
369 98 : if f.is_empty() {
370 10 : Decision::Allow
371 88 : } else if WHOLE_TENANT.contains(&clause) {
372 40 : Decision::Deny(format!(
373 40 : "{clause} acts on everything the tenant holds and cannot be narrowed"
374 40 : ))
375 : } else {
376 48 : Decision::Deny(format!(
377 48 : "{clause} is performed whole or not at all and cannot be narrowed"
378 48 : ))
379 : }
380 : }
381 786 : other => other,
382 : }
383 884 : }
384 :
385 : /// Ask the engine, and fail closed. An engine that panics, or that runs
386 : /// past [`TIMEOUT`], denies: a seam that waved the request through on its
387 : /// own failure would turn a broken addon into an open door.
388 : ///
389 : /// The panic is caught in place rather than on a spawned task, because the
390 : /// task boundary would demand `'static` of the operation, which borrows the
391 : /// request. It is caught TWICE, because an engine has two places to fail:
392 : /// `decide` returns a boxed future, and a synchronous engine does its whole
393 : /// decision in the call that builds that future — the reference engine is
394 : /// `Box::pin(ready(self.judge(..)))` — so guarding only the future guards
395 : /// the half that does nothing.
396 : ///
397 : /// [`TIMEOUT`] can only race the future. Work done before the future exists
398 : /// holds the executor thread, and no timer inside the same task can
399 : /// interrupt it; an engine that blocks is a deployment's own bug, and the
400 : /// bound that catches it is the request timeout in front of the broker.
401 836 : pub async fn decide(engine: &dyn PolicyEngine, subject: &Subject, op: &Operation<'_>) -> Decision {
402 : #[cfg(not(target_arch = "wasm32"))]
403 : {
404 : use futures_util::FutureExt as _;
405 836 : let built =
406 836 : std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| engine.decide(subject, op)));
407 836 : let Ok(fut) = built else {
408 4 : tracing::error!("policy engine {} panicked; denying", engine.name());
409 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic").increment(1);
410 4 : return Decision::Deny(ENGINE_FAILED.to_owned());
411 : };
412 832 : let guarded = std::panic::AssertUnwindSafe(fut).catch_unwind();
413 832 : match tokio::time::timeout(*TIMEOUT, guarded).await {
414 816 : Ok(Ok(d)) => resolve(op.clause, d),
415 : Ok(Err(_)) => {
416 4 : tracing::error!("policy engine {} panicked; denying", engine.name());
417 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic")
418 4 : .increment(1);
419 4 : Decision::Deny(ENGINE_FAILED.to_owned())
420 : }
421 : Err(_) => {
422 12 : tracing::error!(
423 : "policy engine {} did not answer within {:?}; denying",
424 0 : engine.name(),
425 0 : *TIMEOUT
426 : );
427 12 : metrics::counter!("antares_policy_failures_total", "reason" => "timeout")
428 12 : .increment(1);
429 12 : Decision::Deny(ENGINE_TIMED_OUT.to_owned())
430 : }
431 : }
432 : }
433 : // The browser build has no timer to race against and aborts on panic;
434 : // it also loads no addon, so the engine is always the built-in one.
435 : #[cfg(target_arch = "wasm32")]
436 : {
437 : resolve(op.clause, engine.decide(subject, op).await)
438 : }
439 836 : }
440 :
441 : /// Ask the engine about one notification, and fail closed: an engine that
442 : /// panics drops the notification. The document it was handed may already be
443 : /// half-edited, and the broker cannot know which half — so it is not sent.
444 32 : pub fn pre_notify(
445 32 : engine: &dyn PolicyEngine,
446 32 : subject: &Subject,
447 32 : sub: &Value,
448 32 : notification: &mut Value,
449 32 : ) -> NotifyDecision {
450 32 : match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
451 32 : engine.pre_notify(subject, sub, notification)
452 32 : })) {
453 28 : Ok(d) => d,
454 : Err(_) => {
455 4 : tracing::error!(
456 : "policy engine {} panicked on a notification; dropping it",
457 0 : engine.name()
458 : );
459 4 : metrics::counter!("antares_policy_failures_total", "reason" => "panic").increment(1);
460 4 : NotifyDecision::Drop
461 : }
462 : }
463 32 : }
464 :
465 : /// The ProblemDetails `type` and `title` of a refusal. Table 6.3.2-1 names
466 : /// no access-denied error, so none is invented under the ETSI namespace: a
467 : /// refusal is answered with this broker's own URN, documented in the book
468 : /// and noted in the 6.3.2 ledger entry. A client that reads error types can
469 : /// tell a policy refusal from a spec error by the namespace alone.
470 : pub const ACCESS_DENIED_TYPE: &str = "urn:antares:error:AccessDenied";
471 : pub const ACCESS_DENIED_TITLE: &str = "AccessDenied";
472 :
473 : /// The request headers copied into a [`Subject`], comma-separated
474 : /// (`ANTARES_POLICY_SUBJECT_HEADERS`), matched case-insensitively and read
475 : /// once at first use. Empty by default: an engine that wants an identity
476 : /// names the header that carries it, and a broker that copied every header
477 : /// into the subject would be handing an addon the whole request.
478 92 : pub static SUBJECT_HEADERS: LazyLock<Vec<String>> = LazyLock::new(|| {
479 92 : std::env::var("ANTARES_POLICY_SUBJECT_HEADERS")
480 92 : .unwrap_or_default()
481 92 : .split(',')
482 92 : .map(|s| s.trim().to_ascii_lowercase())
483 92 : .filter(|s| !s.is_empty())
484 92 : .collect()
485 92 : });
486 :
487 : /// Who this request is from, as far as the seam is concerned: the tenant it
488 : /// addresses and the headers a deployment named. A header the request does
489 : /// not carry is simply absent — the seam invents nothing.
490 1558 : pub fn subject_of(tenant: &antares_model::TenantId, headers: &HeaderMap) -> Subject {
491 1558 : let mut carried = Vec::new();
492 1558 : for name in SUBJECT_HEADERS.iter() {
493 182 : for value in headers.get_all(name.as_str()) {
494 112 : if let Ok(v) = value.to_str() {
495 112 : carried.push((name.clone(), v.to_owned()));
496 112 : }
497 : }
498 : }
499 1558 : Subject {
500 1558 : tenant: tenant.clone(),
501 1558 : headers: carried,
502 1558 : }
503 1558 : }
504 :
505 : /// The member a Subscription carries its creator's subject in. A
506 : /// broker-internal member like `__context` and `__via`: a client can
507 : /// neither set it nor read it back, it is stripped from every served
508 : /// representation and from the 5.8.1.4 copy forwarded to a Context Source,
509 : /// and 5.2.12 defines no such member — the whole `__` prefix is the
510 : /// broker's, so a member added later inherits every one of those rules
511 : /// instead of a new list to forget.
512 : pub(crate) const SUBJECT_MEMBER: &str = "__subject";
513 :
514 : /// The stored form of a subject: only the headers a deployment named, since
515 : /// the tenant is where the subscription already lives. `None` when the
516 : /// subject carries nothing, so a deployment that named no header stores no
517 : /// member at all.
518 660 : pub(crate) fn subject_member(subject: &Subject) -> Option<Value> {
519 660 : (!subject.headers.is_empty())
520 660 : .then(|| serde_json::to_value(&subject.headers).ok())
521 660 : .flatten()
522 660 : }
523 :
524 : /// The subject a notification is delivered under: the one stored when the
525 : /// subscription was created. A subscription created before the deployment
526 : /// named its headers — or by the broker itself, for the 5.8.1.4 internal
527 : /// copies — simply has none, and the engine is asked about a subject with
528 : /// no headers rather than about the wrong one.
529 230 : pub(crate) fn stored_subject(tenant: &antares_model::TenantId, sub: &Value) -> Subject {
530 : Subject {
531 230 : tenant: tenant.clone(),
532 230 : headers: sub
533 230 : .get(SUBJECT_MEMBER)
534 230 : .cloned()
535 230 : .and_then(|v| serde_json::from_value(v).ok())
536 230 : .unwrap_or_default(),
537 : }
538 230 : }
539 :
540 : /// Remove every broker-internal member from a document about to be served.
541 : ///
542 : /// The `__` prefix is the broker's: the notification `@context` (5.8.6), the
543 : /// 6.3.18 Via chain, a snapshot's synthetic tenant and [`SUBJECT_MEMBER`]
544 : /// all live under it, and no NGSI-LD data type defines a member there. One
545 : /// predicate rather than a list per document type, so a member added later
546 : /// is hidden by construction instead of by remembering every serve point.
547 350 : pub(crate) fn strip_internal(doc: &mut Value) {
548 350 : if let Some(o) = doc.as_object_mut() {
549 2768 : o.retain(|k, _| !k.starts_with("__"));
550 0 : }
551 350 : }
552 :
553 : /// Whether a stored document was made by this subject. Compared on the
554 : /// headers alone: the tenant is where the document already lives, and a
555 : /// deployment that named no header has no subjects to tell apart — every
556 : /// request is the same one, and everything it stored is its own.
557 214 : pub(crate) fn belongs_to(doc: &Value, subject: &Subject) -> bool {
558 214 : stored_subject(&subject.tenant, doc).headers == subject.headers
559 214 : }
560 :
561 : /// A refusal on its way out of a handler. Its own type rather than an
562 : /// `ApiError`, so this module names nothing above it and stays a leaf; the
563 : /// `?` in a handler converts it through `From`.
564 : #[derive(Debug, Clone, PartialEq)]
565 : pub struct Denied(pub String);
566 :
567 : /// The one call a handler makes. Every operation passes through it exactly
568 : /// once, which is what `every_route_asks_the_policy_engine_once` asserts by
569 : /// walking the router with a counting engine.
570 : ///
571 : /// The returned [`Filter`] is what the answer has to be narrowed by; an
572 : /// allow returns the empty filter, which narrows nothing.
573 684 : pub async fn gate(
574 684 : engine: &dyn PolicyEngine,
575 684 : tenant: &antares_model::TenantId,
576 684 : headers: &HeaderMap,
577 684 : op: &Operation<'_>,
578 684 : ) -> Result<Filter, Denied> {
579 684 : match decide(engine, &subject_of(tenant, headers), op).await {
580 572 : Decision::Allow => Ok(Filter::default()),
581 66 : Decision::Filter(f) => Ok(f),
582 46 : Decision::Deny(why) => Err(Denied(why)),
583 : }
584 684 : }
585 :
586 : /// The members of a JSON object, or nothing when it is not one.
587 : #[cfg(any(test, feature = "test-kit"))]
588 52 : fn members(v: &Value) -> Vec<String> {
589 52 : v.as_object()
590 52 : .map(|o| o.keys().cloned().collect())
591 52 : .unwrap_or_default()
592 52 : }
593 :
594 : /// The contract every engine has to hold, so an addon's own tests can call
595 : /// it. It asserts the three things an engine can actually get wrong — it
596 : /// stops answering, it hands back an answer the seam has to override, or it
597 : /// puts something into a notification that was not there — and it asserts
598 : /// them through the seam, so an engine that passes here passes as the
599 : /// broker will call it.
600 : ///
601 : /// The core runs it against [`AllowAll`]; `examples/plugin-example` runs it
602 : /// against the reference engine.
603 : #[cfg(any(test, feature = "test-kit"))]
604 12 : pub async fn run_policy_contract(engine: &dyn PolicyEngine) {
605 12 : let name = engine.name();
606 12 : assert!(!name.is_empty(), "an engine answers to a name");
607 :
608 12 : let subject = Subject {
609 12 : tenant: antares_model::TenantId::default(),
610 12 : headers: vec![("X-Subject".into(), "someone".into())],
611 12 : };
612 12 : let doc = serde_json::json!({
613 12 : "id": "urn:ngsi-ld:Vehicle:1",
614 12 : "type": "Vehicle",
615 12 : "speed": {"type": "Property", "value": 10},
616 12 : "brand": {"type": "Property", "value": "Skoda"}
617 : });
618 :
619 : // The engine answers, and the answer is its own: a deny carrying one of
620 : // the seam's own reasons means it timed out or panicked instead.
621 36 : for clause in ["5.6.1", "5.7.2", "5.8.1"] {
622 36 : match decide(engine, &subject, &Operation::new(clause)).await {
623 24 : Decision::Allow => {}
624 8 : Decision::Deny(why) => assert!(
625 8 : why != ENGINE_TIMED_OUT && why != ENGINE_FAILED,
626 : "{name}: {clause} was denied by the seam, not by the engine: {why}"
627 : ),
628 4 : Decision::Filter(f) => {
629 4 : let mut narrowed = doc.clone();
630 4 : f.project(&mut narrowed);
631 4 : let before = members(&doc);
632 12 : for key in members(&narrowed) {
633 12 : assert!(
634 12 : before.contains(&key),
635 : "{name}: the filter for {clause} added {key:?} to the answer"
636 : );
637 : }
638 : }
639 : }
640 : }
641 :
642 : // An operation the broker performs whole is allowed or refused, never
643 : // done to less than it says: WHOLE_TENANT because there is no narrowed
644 : // form of "delete everything", and the writes because no handler there
645 : // reads a Filter at all (FILTERABLE).
646 96 : for clause in WHOLE_TENANT
647 12 : .iter()
648 12 : .chain(["5.6.1", "5.6.6", "5.8.1", "5.9.2"].iter())
649 : {
650 96 : assert!(
651 96 : !matches!(
652 96 : decide(engine, &subject, &Operation::new(clause)).await,
653 : Decision::Filter(_)
654 : ),
655 : "{name}: {clause} was answered with a filter"
656 : );
657 : }
658 :
659 : // `pre_notify` holds the notification by `&mut`, which is the one place
660 : // an engine can widen rather than narrow: a member it puts there is a
661 : // member no subscriber asked for and no store answered with.
662 12 : let sub = serde_json::json!({"id": "urn:ngsi-ld:Subscription:1", "type": "Subscription"});
663 12 : let mut notification = serde_json::json!({
664 12 : "id": "urn:ngsi-ld:Notification:1",
665 12 : "type": "Notification",
666 12 : "subscriptionId": "urn:ngsi-ld:Subscription:1",
667 12 : "notifiedAt": "2026-01-01T00:00:00Z",
668 12 : "data": [doc.clone()]
669 : });
670 12 : let before = members(¬ification);
671 12 : let entity_members = members(&doc);
672 12 : if pre_notify(engine, &subject, &sub, &mut notification) != NotifyDecision::Drop {
673 56 : for key in members(¬ification) {
674 56 : assert!(
675 56 : before.contains(&key),
676 : "{name}: pre_notify added {key:?} to the notification"
677 : );
678 : }
679 8 : let served = notification
680 8 : .get("data")
681 8 : .and_then(Value::as_array)
682 8 : .cloned()
683 8 : .unwrap_or_default();
684 8 : for entity in &served {
685 28 : for key in members(entity) {
686 28 : assert!(
687 28 : entity_members.contains(&key),
688 : "{name}: pre_notify added {key:?} to a notified Entity"
689 : );
690 : }
691 : }
692 0 : }
693 :
694 : // And the seam stops waiting for this engine, whatever it does.
695 : #[cfg(not(target_arch = "wasm32"))]
696 : {
697 8 : let slow = Slow(engine);
698 8 : assert_eq!(
699 8 : decide(&slow, &subject, &Operation::new("5.6.1")).await,
700 8 : Decision::Deny(ENGINE_TIMED_OUT.to_owned()),
701 : "{name}: the seam waited past its timeout instead of denying"
702 : );
703 : }
704 8 : }
705 :
706 : /// The same engine, one timeout slower: what the contract wraps it in to
707 : /// prove the seam stops waiting.
708 : #[cfg(all(any(test, feature = "test-kit"), not(target_arch = "wasm32")))]
709 : struct Slow<'e>(&'e dyn PolicyEngine);
710 :
711 : #[cfg(all(any(test, feature = "test-kit"), not(target_arch = "wasm32")))]
712 : impl PolicyEngine for Slow<'_> {
713 0 : fn name(&self) -> &str {
714 0 : self.0.name()
715 0 : }
716 :
717 8 : fn decide<'a>(&'a self, subject: &'a Subject, op: &'a Operation<'a>) -> DecisionFuture<'a> {
718 8 : Box::pin(async move {
719 8 : tokio::time::sleep(*TIMEOUT * 2 + Duration::from_millis(50)).await;
720 0 : self.0.decide(subject, op).await
721 0 : })
722 8 : }
723 :
724 0 : fn pre_notify(&self, s: &Subject, sub: &Value, n: &mut Value) -> NotifyDecision {
725 0 : self.0.pre_notify(s, sub, n)
726 0 : }
727 : }
728 :
729 : #[cfg(test)]
730 : mod tests {
731 : use super::*;
732 : use futures_util::FutureExt as _;
733 : use serde_json::json;
734 :
735 20 : fn subject() -> Subject {
736 20 : Subject {
737 20 : tenant: antares_model::TenantId::default(),
738 20 : headers: vec![("X-Subject".into(), "a-token-shaped-string".into())],
739 20 : }
740 20 : }
741 :
742 : /// Run the contract and report whether it refused the engine.
743 8 : async fn contract_holds(engine: &dyn PolicyEngine) -> bool {
744 8 : let hook = std::panic::take_hook();
745 8 : std::panic::set_hook(Box::new(|_| {}));
746 8 : let outcome = std::panic::AssertUnwindSafe(run_policy_contract(engine))
747 8 : .catch_unwind()
748 8 : .await;
749 8 : std::panic::set_hook(hook);
750 8 : outcome.is_ok()
751 8 : }
752 :
753 : #[tokio::test]
754 4 : async fn the_built_in_engine_holds_the_contract() {
755 4 : run_policy_contract(&AllowAll).await;
756 4 : }
757 :
758 : /// An engine that narrows is what the seam is for, and the contract
759 : /// must not stand in its way.
760 : struct Narrowing;
761 :
762 : impl PolicyEngine for Narrowing {
763 4 : fn name(&self) -> &str {
764 4 : "narrowing"
765 4 : }
766 44 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
767 44 : Box::pin(std::future::ready(Decision::Filter(Filter {
768 44 : omit: vec!["brand".into()],
769 44 : restricted: true,
770 44 : ..Filter::default()
771 44 : })))
772 44 : }
773 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, n: &mut Value) -> NotifyDecision {
774 4 : if let Some(data) = n.get_mut("data").and_then(Value::as_array_mut) {
775 4 : for entity in data {
776 4 : if let Some(o) = entity.as_object_mut() {
777 4 : o.remove("brand");
778 4 : }
779 : }
780 0 : }
781 4 : NotifyDecision::Filter(Filter {
782 4 : omit: vec!["brand".into()],
783 4 : ..Filter::default()
784 4 : })
785 4 : }
786 : }
787 :
788 : #[tokio::test]
789 4 : async fn an_engine_that_only_narrows_holds_the_contract() {
790 4 : assert!(contract_holds(&Narrowing).await);
791 4 : }
792 :
793 : /// The contract has teeth or it proves nothing: this engine puts a
794 : /// member into a notification nobody asked for.
795 : struct Widening;
796 :
797 : impl PolicyEngine for Widening {
798 4 : fn name(&self) -> &str {
799 4 : "widening"
800 4 : }
801 44 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
802 44 : Box::pin(std::future::ready(Decision::Allow))
803 44 : }
804 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, n: &mut Value) -> NotifyDecision {
805 4 : if let Some(o) = n.as_object_mut() {
806 4 : o.insert("stowaway".into(), json!(true));
807 4 : }
808 4 : NotifyDecision::Deliver
809 4 : }
810 : }
811 :
812 : #[tokio::test]
813 4 : async fn the_contract_refuses_an_engine_that_widens_a_notification() {
814 4 : assert!(
815 4 : !contract_holds(&Widening).await,
816 4 : "the contract passed an engine that added a member to a notification"
817 4 : );
818 4 : }
819 :
820 : #[test]
821 4 : fn a_pick_keeps_the_frame_and_drops_the_rest() {
822 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle",
823 4 : "speed": 1, "brand": "Skoda"});
824 4 : Filter {
825 4 : pick: vec!["speed".into()],
826 4 : ..Filter::default()
827 4 : }
828 4 : .project(&mut doc);
829 4 : assert_eq!(
830 : doc,
831 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1})
832 : );
833 4 : }
834 :
835 : #[test]
836 4 : fn an_omit_cannot_remove_what_makes_it_an_entity() {
837 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1});
838 4 : Filter {
839 4 : omit: vec!["id".into(), "type".into(), "speed".into()],
840 4 : ..Filter::default()
841 4 : }
842 4 : .project(&mut doc);
843 4 : assert_eq!(
844 : doc,
845 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})
846 : );
847 4 : }
848 :
849 : #[test]
850 4 : fn a_pick_of_something_absent_adds_nothing() {
851 4 : let mut doc = json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": 1});
852 4 : Filter {
853 4 : pick: vec!["mileage".into()],
854 4 : ..Filter::default()
855 4 : }
856 4 : .project(&mut doc);
857 4 : assert_eq!(
858 : doc,
859 4 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle"})
860 : );
861 4 : }
862 :
863 : #[test]
864 4 : fn a_projection_leaves_a_non_object_alone() {
865 4 : let mut doc = json!(["not", "an", "object"]);
866 4 : Filter {
867 4 : pick: vec!["speed".into()],
868 4 : omit: vec!["brand".into()],
869 4 : ..Filter::default()
870 4 : }
871 4 : .project(&mut doc);
872 4 : assert_eq!(doc, json!(["not", "an", "object"]));
873 4 : }
874 :
875 : #[test]
876 4 : fn narrowing_a_whole_tenant_operation_is_a_refusal() {
877 4 : let narrowed = Decision::Filter(Filter {
878 4 : omit: vec!["speed".into()],
879 4 : ..Filter::default()
880 4 : });
881 16 : for clause in WHOLE_TENANT {
882 16 : assert!(matches!(
883 16 : resolve(clause, narrowed.clone()),
884 : Decision::Deny(_)
885 : ));
886 : }
887 : // and every other clause outside FILTERABLE, because the handler
888 : // there never reads the Filter
889 24 : for clause in ["5.6.1", "5.6.6", "5.8.1", "5.9.2", "5.13.2", "5.7.5"] {
890 24 : assert!(
891 24 : matches!(resolve(clause, narrowed.clone()), Decision::Deny(_)),
892 : "{clause} dropped a narrowing instead of refusing it"
893 : );
894 : }
895 4 : }
896 :
897 : /// The reads that do read it keep it: the seam refuses what would be
898 : /// dropped, and nothing more.
899 : #[test]
900 4 : fn a_filterable_read_keeps_its_narrowing() {
901 4 : let narrowed = Decision::Filter(Filter {
902 4 : omit: vec!["speed".into()],
903 4 : ..Filter::default()
904 4 : });
905 24 : for clause in FILTERABLE {
906 24 : assert_eq!(resolve(clause, narrowed.clone()), narrowed, "{clause}");
907 : }
908 4 : }
909 :
910 : #[test]
911 4 : fn an_empty_filter_on_a_whole_tenant_operation_is_not_a_refusal() {
912 4 : assert_eq!(
913 4 : resolve("5.6.21", Decision::Filter(Filter::default())),
914 : Decision::Allow
915 : );
916 4 : }
917 :
918 : struct Panicking;
919 :
920 : impl PolicyEngine for Panicking {
921 0 : fn name(&self) -> &str {
922 0 : "panicking"
923 0 : }
924 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
925 4 : Box::pin(async { panic!("the engine is broken") })
926 4 : }
927 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
928 4 : panic!("the engine is broken")
929 : }
930 : }
931 :
932 : /// The other half of the same failure: a synchronous engine decides in
933 : /// the call that BUILDS the future, so a guard around the future alone
934 : /// never sees it. The reference engine has exactly this shape.
935 : struct PanickingEagerly;
936 :
937 : impl PolicyEngine for PanickingEagerly {
938 0 : fn name(&self) -> &str {
939 0 : "panicking-eagerly"
940 0 : }
941 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
942 4 : panic!("the engine is broken")
943 : }
944 0 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
945 0 : NotifyDecision::Deliver
946 0 : }
947 : }
948 :
949 : #[tokio::test]
950 4 : async fn an_engine_that_panics_before_it_returns_a_future_denies_too() {
951 4 : let hook = std::panic::take_hook();
952 4 : std::panic::set_hook(Box::new(|_| {}));
953 4 : let decided = decide(&PanickingEagerly, &subject(), &Operation::new("5.6.1")).await;
954 4 : std::panic::set_hook(hook);
955 4 : assert_eq!(decided, Decision::Deny(ENGINE_FAILED.to_owned()));
956 4 : }
957 :
958 : #[tokio::test]
959 4 : async fn an_engine_that_panics_denies_rather_than_allows() {
960 4 : let hook = std::panic::take_hook();
961 8 : std::panic::set_hook(Box::new(|_| {}));
962 4 : let decided = decide(&Panicking, &subject(), &Operation::new("5.6.1")).await;
963 4 : let mut notification = json!({"type": "Notification"});
964 4 : let notified = pre_notify(&Panicking, &subject(), &json!({}), &mut notification);
965 4 : std::panic::set_hook(hook);
966 4 : assert_eq!(decided, Decision::Deny(ENGINE_FAILED.to_owned()));
967 4 : assert_eq!(notified, NotifyDecision::Drop);
968 4 : }
969 :
970 : /// Real time, not a paused clock: the workspace tokio is built without
971 : /// `test-util`, so the test pays the timeout it asserts.
972 : #[tokio::test]
973 4 : async fn an_engine_that_never_answers_denies() {
974 : struct Never;
975 4 : impl PolicyEngine for Never {
976 4 : fn name(&self) -> &str {
977 4 : "never"
978 4 : }
979 4 : fn decide<'a>(&'a self, _s: &'a Subject, _o: &'a Operation<'a>) -> DecisionFuture<'a> {
980 4 : Box::pin(std::future::pending())
981 4 : }
982 4 : fn pre_notify(&self, _s: &Subject, _sub: &Value, _n: &mut Value) -> NotifyDecision {
983 4 : NotifyDecision::Deliver
984 4 : }
985 4 : }
986 4 : assert_eq!(
987 4 : decide(&Never, &subject(), &Operation::new("5.6.1")).await,
988 4 : Decision::Deny(ENGINE_TIMED_OUT.to_owned())
989 4 : );
990 4 : }
991 :
992 : #[test]
993 4 : fn the_subject_never_prints_its_header_values() {
994 4 : let printed = format!("{:?}", subject());
995 4 : assert!(printed.contains("X-Subject"), "{printed}");
996 4 : assert!(
997 4 : !printed.contains("a-token-shaped-string"),
998 : "a header value reached a log line: {printed}"
999 : );
1000 4 : }
1001 :
1002 : #[test]
1003 4 : fn an_operation_never_prints_the_payload() {
1004 4 : let body = json!({"id": "urn:ngsi-ld:Vehicle:1", "plate": "BB-123-XY"});
1005 4 : let printed = format!(
1006 : "{:?}",
1007 4 : Operation {
1008 4 : body: Some(&body),
1009 4 : ..Operation::new("5.6.1")
1010 4 : }
1011 : );
1012 4 : assert!(!printed.contains("BB-123-XY"), "{printed}");
1013 4 : }
1014 : }
|
