From 3d83e63e8c21fb7c11aa5e029b0de8dbd81407f2 Mon Sep 17 00:00:00 2001 From: dusan Date: Thu, 6 Aug 2026 18:58:49 +0200 Subject: [PATCH 1/6] Serve the broker auth callout directly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Atom implements FluxMQ's fluxmq.auth.v1.AuthService, so a broker can authenticate clients and authorize topics against Atom with no adapter service in between. The proto is vendored verbatim from FluxMQ. Its package line is part of the wire contract — the path a broker dials is derived from it — so it must not be renamed. How a topic names an object is configuration (ATOM_BROKER_TOPIC_TEMPLATE), which keeps any particular deployment's topic vocabulary out of Atom; an adapter service is still the right answer where the mapping needs more than a grammar. Two invariants the implementation rests on: - Denials are answers, not errors. Every rejection returns a successful RPC carrying a false verdict; only infrastructure failure returns a gRPC error. A broker wraps this callout in a circuit breaker, and a tripped breaker rejects every client connection, so one device retrying a stale password must not be able to take the broker's whole auth path down. Rate limiting is on that list because it is the failure a bad client can trigger at will. - Tenant comes from the subject, not from config or the topic. Authenticate resolves the identifier across tenants and the entity's own tenant comes back with it, so the zero-configuration case needs no tenant in the topic and no username grammar. A {tenant} template segment only scopes alias resolution; it is deliberately not checked against the subject's tenant, because cross-tenant grants are legitimate and that call belongs to the PDP. Off by default. It is the only gRPC service here with no bearer token to check, so it authenticates its caller at the transport via the listener's mTLS client CA; mounted on a plaintext listener, anything that can reach the port could authenticate and authorize as any principal. A startup warning fires if it is enabled without a client CA. --- AGENTS.md | 53 ++++ build.rs | 5 +- proto/broker/v1/auth.proto | 151 ++++++++++ src/broker_auth/mod.rs | 23 ++ src/broker_auth/service.rs | 364 ++++++++++++++++++++++ src/broker_auth/topic.rs | 502 +++++++++++++++++++++++++++++++ src/config.rs | 110 +++++++ src/grpc.rs | 44 ++- src/lib.rs | 1 + tests/m29_broker_auth_callout.rs | 405 +++++++++++++++++++++++++ 10 files changed, 1655 insertions(+), 3 deletions(-) create mode 100644 proto/broker/v1/auth.proto create mode 100644 src/broker_auth/mod.rs create mode 100644 src/broker_auth/service.rs create mode 100644 src/broker_auth/topic.rs create mode 100644 tests/m29_broker_auth_callout.rs diff --git a/AGENTS.md b/AGENTS.md index 6db146d6..23477160 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,6 +30,10 @@ src/ │ RequireManage extractor + has_global_manage() helper keys.rs — ES256 signing keys (primary/standby/retired), encryption at rest grpc.rs — Tonic services: AuthService, AuthzService.Check, CertificateService + broker_auth/ — the broker auth callout: Atom serving FluxMQ's + │ `fluxmq.auth.v1.AuthService` directly (off by default) + │ topic.rs — the configurable topic→object grammar + │ service.rs — Authenticate/Authorize over the existing credential + PDP paths graphql/ — schema + per-domain resolvers (the live admin/API surface) db.rs — pool creation (configurable pool) models/ @@ -221,6 +225,55 @@ must then be confined to a private network or a service mesh that provides transport security, and a startup warning is logged. (The HTTP rate limiter does not cover gRPC; see backlog #10.) +### Broker auth callout + +A message broker delegates connect-time credential checks and per-topic access +control to an external gRPC service. Atom implements that contract itself +(`src/broker_auth/`), so a broker can be pointed straight at Atom with **no +adapter service in between**. The proto is vendored verbatim from FluxMQ at +`proto/broker/v1/auth.proto`; its `package fluxmq.auth.v1` line is part of the +wire contract (the dialled path is `/fluxmq.auth.v1.AuthService/Authorize`) and +must not be renamed. Check for drift with a `diff` against the FluxMQ checkout. + +Config (`ATOM_BROKER_*`), all optional: + +| Variable | Default | Meaning | +|---|---|---| +| `ATOM_BROKER_AUTH_ENABLED` | `false` | mount the callout | +| `ATOM_BROKER_TOPIC_TEMPLATE` | `{resource}/#` | comma-separated templates, tried in order | +| `ATOM_BROKER_TOPIC_REF` | `alias` | `alias` or `uuid` — how a bound segment names an object | +| `ATOM_BROKER_CREDENTIAL_KIND` | `password` | `password` or `shared_key` | + +**Off by default for a security reason, not a rollout one.** It is the only gRPC +service here with no bearer token to check — a broker's callout client cannot +send one — so it authenticates its caller at the transport, via +`ATOM_GRPC_TLS_CLIENT_CA_PATH`. Mounted on a plaintext listener, anything that +can reach the port can authenticate and authorize as any principal. Enable it +with a client CA that signs brokers and nothing else. A startup warning fires if +it is enabled without one. + +Two invariants worth not relearning: + +- **Denials are answers, not errors.** Every rejection — bad password, unknown + entity, unparseable topic, policy deny, *rate limit* — returns a successful RPC + carrying a false verdict. Only infrastructure failure returns a gRPC error. A + broker wraps this callout in a circuit breaker, and a tripped breaker rejects + **every** client connection; one device retrying a stale password must not be + able to take the broker's whole auth path down. Rate limiting is on that list + because it is the failure a bad client can trigger at will. +- **Tenant comes from the subject, not from config or the topic.** Authenticate + resolves the identifier across tenants and the entity's own tenant comes back + with it, so the zero-configuration case needs no tenant in the topic and no + username grammar. A `{tenant}` template segment, when present, only scopes + alias resolution — it is deliberately **not** checked against the subject's + tenant, because cross-tenant grants are legitimate and that call belongs to + the PDP, not to a hardcoded equality test. + +An adapter service is still right where the mapping needs more than a grammar — +route resolution, multi-service composition. Both speak the same wire contract, +so a deployment picks one by pointing the broker's `auth.external.url` at Atom +or at the adapter. + ## Metrics Prometheus metrics are exposed at `GET /metrics` (text exposition). All metric diff --git a/build.rs b/build.rs index b2a2ce5f..b8464a9e 100644 --- a/build.rs +++ b/build.rs @@ -7,7 +7,10 @@ fn main() -> Result<(), Box> { println!("cargo:rustc-env=ATOM_VERSION={version}"); println!("cargo:rustc-env=ATOM_REVISION={revision}"); - tonic_build::compile_protos("proto/atom/v1/atom.proto")?; + tonic_build::configure().compile_protos( + &["proto/atom/v1/atom.proto", "proto/broker/v1/auth.proto"], + &["proto"], + )?; Ok(()) } diff --git a/proto/broker/v1/auth.proto b/proto/broker/v1/auth.proto new file mode 100644 index 00000000..8f62a903 --- /dev/null +++ b/proto/broker/v1/auth.proto @@ -0,0 +1,151 @@ +// Copyright (c) Abstract Machines +// SPDX-License-Identifier: Apache-2.0 + +// Vendored verbatim from FluxMQ `proto/auth/v1/auth.proto`. Atom implements +// `AuthService` so a broker can call it directly, with no adapter service in +// between. Check for drift with: +// +// diff proto/broker/v1/auth.proto \ +// $FLUXMQ/proto/auth/v1/auth.proto +// +// The `package` line is part of the wire contract — the gRPC path a broker +// dials is `/fluxmq.auth.v1.AuthService/Authorize`, derived from the proto +// package plus service name. Renaming it here silently stops matching. +// `HookService` is vendored for completeness but is not implemented. + +syntax = "proto3"; + +package fluxmq.auth.v1; + +// AuthService is a callout service for external authentication and +// authorization. Broker implementations call this service during connection +// establishment, publish, and subscribe to delegate credential validation +// and topic-level access control to an external provider. +service AuthService { + // Authenticate validates client credentials presented during connection. + // The server resolves the credentials to an external identity (e.g. a UUID) + // which the broker stores on the session and passes in subsequent Authorize + // calls. + rpc Authenticate(AuthnReq) returns (AuthnRes); + + // Authorize checks whether a previously authenticated identity is allowed + // to perform a given action on the effective topic/filter. When the optional + // topic normalizer is configured, the broker authorizes the normalized value. + rpc Authorize(AuthzReq) returns (AuthzRes); +} + +// HookService is an optional callout service for blocking broker hooks. +// Hook handlers may allow, deny, or mutate only the fields supported by the +// hook point before the broker continues processing. +service HookService { + rpc Handle(HookReq) returns (HookRes); +} + +// Protocol identifies which messaging protocol the client connected with. +enum Protocol { + Unspecified = 0; + MQTT = 1; + AMQP_1_0 = 2; + AMQP_0_9_1 = 3; + HTTP = 4; + CoAP = 5; +} + +message AuthnReq { + // Client identifier from the connection handshake. + string client_id = 1; + // Username credential. + string username = 2; + // Password credential. + string password = 3; + // Protocol the client connected with. + Protocol protocol = 4; +} + +message AuthnRes { + // Whether the credentials are valid. + bool authenticated = 1; + // External identity resolved by the auth provider (e.g. a client UUID). + // The broker stores this on the session and passes it in AuthzReq. + string id = 2; + // Machine-readable reason code (0 = success). + uint32 reason_code = 3; + // Human-readable reason for rejection (empty on success). + string reason = 4; +} + +message AuthzReq { + // External identity returned from AuthnRes.id. + string external_id = 1; + // Raw topic or topic filter from the publish/subscribe operation. + string topic = 2; + // The action being requested. + Action action = 3; +} + +enum Action { + None = 0; + Publish = 1; + Subscribe = 2; +} + +message AuthzRes { + // Whether the action is authorized. + bool authorized = 1; + // Machine-readable reason code (0 = authorized). + uint32 reason_code = 2; + // Human-readable reason for rejection (empty on success). + string reason = 3; +} + +enum HookType { + HookTypeUnspecified = 0; + AuthOnRegister = 1; + AuthOnPublish = 2; + AuthOnSubscribe = 3; + AuthOnUnsubscribe = 4; +} + +enum HookResult { + HookResultUnspecified = 0; + HookResultOk = 1; + HookResultDeny = 2; +} + +message HookReq { + HookType hook = 1; + // Protocol-level client identifier. + string client_id = 2; + // External identity returned from AuthnRes.id when available. + string external_id = 3; + // Protocol the client connected with. + Protocol protocol = 4; + // Topic, topic filter, or address being handled. + string topic = 5; + // Publish payload. Empty for non-publish hooks. + bytes payload = 6; + uint32 qos = 7; + bool retain = 8; + map properties = 9; + // Username/password are populated only for register/authenticate hooks. + string username = 10; + string password = 11; +} + +message HookRes { + HookResult result = 1; + // Empty means keep the requested topic/filter. + string topic = 2; + bytes payload = 3; + bool payload_set = 4; + uint32 qos = 5; + bool qos_set = 6; + bool retain = 7; + bool retain_set = 8; + // Properties are merged into the current properties map. + map properties = 9; + uint32 reason_code = 10; + string reason = 11; + // AuthOnRegister may set or override the external identity. + string external_id = 12; +} diff --git a/src/broker_auth/mod.rs b/src/broker_auth/mod.rs new file mode 100644 index 00000000..dc791e17 --- /dev/null +++ b/src/broker_auth/mod.rs @@ -0,0 +1,23 @@ +//! Broker auth callout — Atom implementing FluxMQ's `AuthService` directly. +//! +//! A message broker delegates connect-time credential checks and per-topic +//! access control to an external service over gRPC. Atom serves that contract +//! itself, so a deployment can point a broker straight at Atom with no adapter +//! service in between. +//! +//! The contract is deliberately the broker's, not Atom's: the wire path a +//! broker dials is `/fluxmq.auth.v1.AuthService/...`, derived from the vendored +//! proto's package. Everything Atom needs beyond that — how a topic names an +//! object — is configuration, so Atom never learns a particular deployment's +//! topic vocabulary. See [`topic`] for the grammar. +//! +//! An adapter service is still the right answer where the mapping needs more +//! than a grammar (route resolution, multi-service composition). Both speak the +//! same wire contract, so a deployment picks one by pointing the broker's +//! `auth.external.url` at Atom or at the adapter. + +pub mod service; +pub mod topic; + +pub use service::BrokerAuth; +pub use topic::{TopicMatch, TopicTemplate, TopicTemplateSet}; diff --git a/src/broker_auth/service.rs b/src/broker_auth/service.rs new file mode 100644 index 00000000..82fc8615 --- /dev/null +++ b/src/broker_auth/service.rs @@ -0,0 +1,364 @@ +//! The `fluxmq.auth.v1.AuthService` implementation. +//! +//! ## Denials are answers, not errors +//! +//! Every rejection this service can reach — bad password, unknown entity, +//! unparseable topic, policy deny, rate limit — returns a *successful* RPC +//! carrying `authenticated: false` / `authorized: false`. Only a genuine +//! infrastructure failure returns a gRPC error. +//! +//! That is not stylistic. A broker wraps this callout in a circuit breaker; a +//! run of RPC errors trips it, and a tripped breaker rejects **every** client +//! connection, not just the one that misbehaved. A single device retrying with +//! a stale password must not be able to take the broker's whole auth path down. +//! Rate limiting is on that list for the same reason: it is the one failure a +//! bad client can trigger at will. + +use tonic::{Request, Response, Status}; +use uuid::Uuid; + +use crate::{ + audit, + authz::{engine, repo}, + certs, + config::BrokerTopicRef, + error::AppError, + identity::service as identity_service, + models::{alias::AliasObjectClass, enums::AuditOutcome, policy::AuthzRequest}, + state::AppState, +}; + +use super::topic::TopicMatch; + +// Generated from the vendored proto/broker/v1/auth.proto. The module path is +// the proto package, which is also the gRPC wire path a broker dials. +pub mod proto { + tonic::include_proto!("fluxmq.auth.v1"); +} + +pub use proto::auth_service_server::AuthServiceServer as BrokerAuthServiceServer; +use proto::{auth_service_server::AuthService, Action, AuthnReq, AuthnRes, AuthzReq, AuthzRes}; + +/// MQTT v5 reason codes, reused so a broker can forward something meaningful in +/// its CONNACK/SUBACK rather than a bare boolean. +const REASON_SUCCESS: u32 = 0x00; +const REASON_BAD_CREDENTIALS: u32 = 0x86; +const REASON_NOT_AUTHORIZED: u32 = 0x87; + +/// The object kind every broker topic addresses. Topics name channels, which +/// are `resources` in Atom's model. +const OBJECT_KIND: &str = "resource"; + +pub struct BrokerAuth { + state: AppState, +} + +impl BrokerAuth { + pub fn new(state: AppState) -> Self { + Self { state } + } +} + +/// Whether a failure is a decision about the request rather than a fault in +/// Atom. See the module docs: only the latter may surface as a gRPC error. +fn is_decision(err: &AppError) -> bool { + matches!( + err, + AppError::NotFound(_) + | AppError::BadRequest(_) + | AppError::Unauthorized(_) + | AppError::Forbidden + | AppError::Conflict(_) + | AppError::RateLimited { .. } + ) +} + +fn authn_denied(reason: impl Into) -> AuthnRes { + AuthnRes { + authenticated: false, + id: String::new(), + reason_code: REASON_BAD_CREDENTIALS, + reason: reason.into(), + } +} + +fn authz_denied(reason: impl Into) -> AuthzRes { + AuthzRes { + authorized: false, + reason_code: REASON_NOT_AUTHORIZED, + reason: reason.into(), + } +} + +#[tonic::async_trait] +impl AuthService for BrokerAuth { + /// Resolve broker credentials to an Atom entity. + /// + /// The caller is trusted at the transport (the gRPC listener's mTLS client + /// CA); there is no bearer token on this path, because a broker's callout + /// client has nowhere to put one. + async fn authenticate(&self, request: Request) -> Result, Status> { + let req = request.into_inner(); + + // Short-circuit before the database and, importantly, before the login + // rate limiter: a flood of anonymous connects would otherwise spend + // Atom's throttle budget and start returning errors that trip the + // broker's circuit breaker. + if req.username.is_empty() || req.password.is_empty() { + return Ok(Response::new(authn_denied("missing credentials"))); + } + + // No tenant selector: the identifier is resolved across tenants and the + // entity's own tenant comes back with it. That is what lets this work + // with no configuration in a multi-tenant deployment. + let result = identity_service::authenticate_credential_in_tenant( + &self.state.pool, + &self.state.config, + &req.username, + &req.password, + None, + self.state.config.broker_auth.credential_kind, + ) + .await; + + match result { + Ok(authenticated) => { + tracing::debug!( + client_id = %req.client_id, + entity_id = %authenticated.entity_id, + "broker authenticate: allow" + ); + Ok(Response::new(AuthnRes { + authenticated: true, + id: authenticated.entity_id.to_string(), + reason_code: REASON_SUCCESS, + reason: String::new(), + })) + } + Err(err) if is_decision(&err) => { + tracing::debug!( + client_id = %req.client_id, + error = %err, + "broker authenticate: deny" + ); + Ok(Response::new(authn_denied(err.to_string()))) + } + Err(err) => { + tracing::error!(client_id = %req.client_id, error = %err, "broker authenticate: failed"); + Err(Status::from(err)) + } + } + } + + /// Decide one publish or subscribe against the topic's object. + async fn authorize(&self, request: Request) -> Result, Status> { + let req = request.into_inner(); + + let Some(action) = action_name(req.action) else { + return Ok(Response::new(authz_denied("unsupported action"))); + }; + + // `external_id` is whatever Authenticate returned. If the broker did not + // authenticate this client it passes the protocol-level client id, which + // is not an Atom subject — deny rather than guess. + let Ok(subject_id) = Uuid::parse_str(&req.external_id) else { + return Ok(Response::new(authz_denied("subject is not an Atom entity"))); + }; + + let Some(matched) = self + .state + .config + .broker_auth + .topic_templates + .match_topic(&req.topic) + else { + return Ok(Response::new(authz_denied( + "topic does not address one object", + ))); + }; + + let object_id = match self.resolve_object(subject_id, &matched).await { + Ok(object_id) => object_id, + Err(err) if is_decision(&err) => { + tracing::debug!( + subject_id = %subject_id, topic = %req.topic, error = %err, + "broker authorize: deny (unresolved object)" + ); + return Ok(Response::new(authz_denied(err.to_string()))); + } + Err(err) => { + tracing::error!(subject_id = %subject_id, topic = %req.topic, error = %err, + "broker authorize: failed"); + return Err(Status::from(err)); + } + }; + + let authz_req = AuthzRequest { + subject_id, + action: action.to_string(), + resource_id: None, + object_kind: Some(OBJECT_KIND.to_string()), + object_id: Some(object_id), + context: serde_json::json!({ + "topic": req.topic, + "subtopic": matched.subtopic, + "connection": action, + }), + }; + + // No ceiling: the broker is not acting under a scoped access token, and + // the subject's own grants are the whole authority here. + let decision = match engine::evaluate_with_ceiling(&self.state.pool, &authz_req, None).await + { + Ok(decision) => decision, + Err(err) if is_decision(&err) => { + return Ok(Response::new(authz_denied(err.to_string()))) + } + Err(err) => { + tracing::error!(subject_id = %subject_id, topic = %req.topic, error = %err, + "broker authorize: evaluation failed"); + return Err(Status::from(err)); + } + }; + + let tenant_id = matched + .tenant + .as_deref() + .and_then(|tenant| Uuid::parse_str(tenant).ok()); + audit::write_hot_path( + &self.state.pool, + self.state.config.audit_policy, + self.state.config.events.enabled(), + audit::HotPathAuditKind::AuthzCheck, + audit::AuditEvent { + actor_entity_id: Some(subject_id), + tenant_id, + target_kind: Some(OBJECT_KIND), + target_id: Some(object_id), + event: "authz.check", + outcome: if decision.allowed { + AuditOutcome::Allow + } else { + AuditOutcome::Deny + }, + details: serde_json::json!({ + "subject_id": subject_id, + "action": action, + "object_id": object_id, + "topic": req.topic, + "transport": "grpc:broker", + }), + }, + ) + .await; + + Ok(Response::new(if decision.allowed { + AuthzRes { + authorized: true, + reason_code: REASON_SUCCESS, + reason: String::new(), + } + } else { + authz_denied(decision.reason) + })) + } +} + +impl BrokerAuth { + /// Turn the bound topic segments into an object UUID. + /// + /// When the template carries no `{tenant}`, the subject's own tenant is the + /// resolution scope — which is why the zero-configuration case needs no + /// tenant in the topic at all. + async fn resolve_object( + &self, + subject_id: Uuid, + matched: &TopicMatch, + ) -> Result { + match self.state.config.broker_auth.topic_ref { + BrokerTopicRef::Uuid => Uuid::parse_str(&matched.resource) + .map_err(|_| AppError::bad_request("topic object segment is not a UUID")), + BrokerTopicRef::Alias => { + let (tenant_id, tenant_alias, global) = match matched.tenant.as_deref() { + // A `{tenant}` segment scopes resolution on its own; it is + // not checked against the subject's tenant. Cross-tenant + // grants are legitimate, so that call belongs to the PDP, + // not to a hardcoded equality test here. + Some(alias) => (None, Some(alias), false), + None => { + match certs::repo::entity_tenant_id(&self.state.pool, subject_id).await? { + Some(tenant_id) => (Some(tenant_id), None, false), + // A tenantless subject addresses tenantless objects. + None => (None, None, true), + } + } + }; + + let resolved = repo::resolve_alias( + &self.state.pool, + tenant_id, + tenant_alias, + global, + AliasObjectClass::Resource, + &matched.resource, + ) + .await?; + Ok(resolved.object_id) + } + } + } +} + +fn action_name(action: i32) -> Option<&'static str> { + match Action::try_from(action) { + Ok(Action::Publish) => Some("publish"), + Ok(Action::Subscribe) => Some("subscribe"), + Ok(Action::None) | Err(_) => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn actions_map_to_atom_capability_names() { + assert_eq!(action_name(Action::Publish as i32), Some("publish")); + assert_eq!(action_name(Action::Subscribe as i32), Some("subscribe")); + } + + #[test] + fn unset_and_unknown_actions_are_rejected() { + assert_eq!(action_name(Action::None as i32), None); + assert_eq!(action_name(99), None); + } + + #[test] + fn request_shaped_failures_are_decisions_not_faults() { + assert!(is_decision(&AppError::not_found("no such resource"))); + assert!(is_decision(&AppError::unauthorized("invalid credentials"))); + assert!(is_decision(&AppError::Forbidden)); + assert!(is_decision(&AppError::bad_request("bad topic"))); + // A client can trigger this at will; letting it surface as an RPC error + // would let one bad device trip the broker's circuit breaker. + assert!(is_decision(&AppError::RateLimited { + message: "slow down".into(), + retry_after_secs: 1, + })); + } + + #[test] + fn infrastructure_failures_are_faults() { + assert!(!is_decision(&AppError::Internal(anyhow::anyhow!("boom")))); + assert!(!is_decision(&AppError::Database(sqlx::Error::PoolClosed))); + } + + #[test] + fn denial_responses_carry_the_not_authorized_reason_code() { + assert_eq!(authz_denied("nope").reason_code, REASON_NOT_AUTHORIZED); + assert!(!authz_denied("nope").authorized); + assert_eq!(authn_denied("nope").reason_code, REASON_BAD_CREDENTIALS); + assert!(!authn_denied("nope").authenticated); + assert!(authn_denied("nope").id.is_empty()); + } +} diff --git a/src/broker_auth/topic.rs b/src/broker_auth/topic.rs new file mode 100644 index 00000000..9869c11f --- /dev/null +++ b/src/broker_auth/topic.rs @@ -0,0 +1,502 @@ +//! Topic template matching for the broker auth callout. +//! +//! A broker hands Atom a raw topic (`PUBLISH`) or topic filter (`SUBSCRIBE`). +//! Atom needs two things out of it: which tenant to resolve in, and which +//! object is being addressed. The mapping is deployment-specific — a plain +//! broker uses the first segment, Magistrala uses `m/{tenant}/c/{resource}` — +//! so the grammar is configuration, not code. Atom learns "topics bind +//! segments to a tenant and an object", never any particular topic layout. +//! +//! Template grammar, segments separated by `/`: +//! +//! | Token | Meaning | +//! |---------------|------------------------------------------------------| +//! | `{tenant}` | optional; binds the resolution tenant | +//! | `{resource}` | required; binds the addressed object, one segment | +//! | `{subtopic}` | optional; binds the remainder, passed as PDP context | +//! | `+` | matches one segment, discarded | +//! | `#` | matches zero or more segments, terminal, discarded | +//! | anything else | literal, must match exactly | +//! +//! `{tenant}` is a *resolution scope*, not a guard: it is not required to equal +//! the subject's tenant. Atom supports cross-tenant grants, so forcing equality +//! here would deny legitimate access from a hardcoded rule instead of from the +//! PDP. Resolve in the named tenant and let the engine decide. + +use std::fmt; + +/// One parsed template segment. +#[derive(Debug, Clone, PartialEq, Eq)] +enum Segment { + Literal(String), + Tenant, + Resource, + /// Binds the remainder of the topic. Terminal. + Subtopic, + /// Matches exactly one segment, discarded. + SingleWild, + /// Matches zero or more remaining segments, discarded. Terminal. + MultiWild, +} + +impl Segment { + /// True when the segment pins a specific value at its position — either a + /// literal that must match or a placeholder that must bind a usable name. + /// A broker wildcard reaching one of these makes the request unresolvable. + fn is_pinned(&self) -> bool { + matches!( + self, + Segment::Literal(_) | Segment::Tenant | Segment::Resource + ) + } + + /// True when the segment consumes the rest of the topic and ends matching. + fn is_terminal(&self) -> bool { + matches!(self, Segment::Subtopic | Segment::MultiWild) + } +} + +/// A parsed, validated topic template. Built once at startup; a malformed +/// template fails the process rather than silently denying every request. +#[derive(Debug, Clone)] +pub struct TopicTemplate { + segments: Vec, + raw: String, +} + +/// What a template extracted from one topic. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TopicMatch { + /// Present only when the template carries `{tenant}`. `None` means "use the + /// subject's own tenant". + pub tenant: Option, + pub resource: String, + /// The remainder bound by `{subtopic}`, if the template has one. Empty + /// remainder yields `None`. + pub subtopic: Option, +} + +/// Why a template string could not be parsed. Startup-time only. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TemplateParseError { + template: String, + reason: String, +} + +impl fmt::Display for TemplateParseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "invalid topic template {:?}: {}", + self.template, self.reason + ) + } +} + +impl std::error::Error for TemplateParseError {} + +impl TopicTemplate { + pub fn parse(template: &str) -> Result { + let raw = template.trim().to_string(); + let fail = |reason: &str| TemplateParseError { + template: raw.clone(), + reason: reason.to_string(), + }; + + if raw.is_empty() { + return Err(fail("template must not be empty")); + } + + let mut segments = Vec::new(); + for token in raw.split('/') { + let segment = match token { + "{tenant}" => Segment::Tenant, + "{resource}" => Segment::Resource, + "{subtopic}" => Segment::Subtopic, + "+" => Segment::SingleWild, + "#" => Segment::MultiWild, + other => { + if other.starts_with('{') || other.ends_with('}') { + return Err(fail(&format!( + "unknown placeholder {other:?} \ + (expected {{tenant}}, {{resource}} or {{subtopic}})" + ))); + } + if other.contains('+') || other.contains('#') { + return Err(fail(&format!( + "literal segment {other:?} must not contain '+' or '#'" + ))); + } + Segment::Literal(other.to_string()) + } + }; + segments.push(segment); + } + + let count = |wanted: &Segment| segments.iter().filter(|seg| *seg == wanted).count(); + if count(&Segment::Resource) != 1 { + return Err(fail("template must contain exactly one {resource}")); + } + if count(&Segment::Tenant) > 1 { + return Err(fail("template must contain at most one {tenant}")); + } + if count(&Segment::Subtopic) > 1 { + return Err(fail("template must contain at most one {subtopic}")); + } + + // Terminal segments consume the remainder, so anything after them is + // unreachable and almost certainly a mistake in the operator's config. + if let Some(index) = segments.iter().position(Segment::is_terminal) { + if index != segments.len() - 1 { + return Err(fail( + "'#' and {subtopic} consume the rest of the topic and must come last", + )); + } + } + + Ok(Self { segments, raw }) + } + + /// The template as configured, for diagnostics. + pub fn as_str(&self) -> &str { + &self.raw + } + + /// Match one topic. `None` means this template does not apply, or applies + /// but cannot yield a usable binding — the caller denies either way. + /// + /// `topic` may be a concrete publish topic or a subscribe filter containing + /// `+`/`#`. A broker wildcard is only acceptable where the template does not + /// pin the position: a filter spanning many resources cannot be authorized + /// as one object. + pub fn match_topic(&self, topic: &str) -> Option { + let topic = topic.strip_prefix('/').unwrap_or(topic); + if topic.is_empty() { + return None; + } + let tokens: Vec<&str> = topic.split('/').collect(); + + // MQTT allows '#' only as the final segment. Anything else is a + // malformed filter, not something to interpret generously. + if let Some(index) = tokens.iter().position(|token| *token == "#") { + if index != tokens.len() - 1 { + return None; + } + } + + // Every non-terminal template segment consumes exactly one token, so + // template and topic indices line up until a terminal segment. That + // makes the wildcard check a straight positional comparison: a broker + // '#' at index i leaves every position from i onward unconstrained, so + // no pinned segment may sit there. + if let Some(hash) = tokens.iter().position(|token| *token == "#") { + if self + .segments + .iter() + .skip(hash) + .any(|segment| segment.is_pinned()) + { + return None; + } + } + + let mut tenant = None; + let mut resource = None; + let mut subtopic = None; + let mut cursor = 0usize; + + for segment in &self.segments { + match segment { + Segment::MultiWild => { + cursor = tokens.len(); + break; + } + Segment::Subtopic => { + let rest = tokens[cursor.min(tokens.len())..].join("/"); + subtopic = (!rest.is_empty()).then_some(rest); + cursor = tokens.len(); + break; + } + _ => { + let token = *tokens.get(cursor)?; + match segment { + Segment::Literal(literal) => { + if token != literal { + return None; + } + } + Segment::Tenant | Segment::Resource => { + // Already excluded '#' positionally above; '+' is + // still possible and is equally unresolvable. + if token == "+" || token.is_empty() { + return None; + } + if matches!(segment, Segment::Tenant) { + tenant = Some(token.to_string()); + } else { + resource = Some(token.to_string()); + } + } + Segment::SingleWild => {} + Segment::Subtopic | Segment::MultiWild => unreachable!("handled above"), + } + cursor += 1; + } + } + } + + // Without a terminal segment the template must consume the whole topic; + // a longer topic addresses something the template does not describe. + if cursor != tokens.len() { + return None; + } + + Some(TopicMatch { + tenant, + resource: resource?, + subtopic, + }) + } +} + +/// The configured templates, tried in order. First one that yields a binding +/// wins; if none do, the caller denies. +#[derive(Debug, Clone)] +pub struct TopicTemplateSet { + templates: Vec, +} + +impl TopicTemplateSet { + pub fn parse_list(templates: &[String]) -> Result { + let templates = templates + .iter() + .map(|template| TopicTemplate::parse(template)) + .collect::, _>>()?; + Ok(Self { templates }) + } + + pub fn match_topic(&self, topic: &str) -> Option { + self.templates + .iter() + .find_map(|template| template.match_topic(topic)) + } + + pub fn iter(&self) -> impl Iterator { + self.templates.iter() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse(template: &str) -> TopicTemplate { + TopicTemplate::parse(template).expect("template should parse") + } + + fn matched(template: &str, topic: &str) -> TopicMatch { + parse(template) + .match_topic(topic) + .unwrap_or_else(|| panic!("{template:?} should match {topic:?}")) + } + + fn denied(template: &str, topic: &str) { + assert!( + parse(template).match_topic(topic).is_none(), + "{template:?} should not match {topic:?}" + ); + } + + #[test] + fn default_template_binds_the_first_segment() { + let result = matched("{resource}/#", "sensors/eu/temp"); + assert_eq!(result.resource, "sensors"); + assert_eq!(result.tenant, None); + assert_eq!(result.subtopic, None); + } + + #[test] + fn default_template_matches_a_bare_resource() { + assert_eq!(matched("{resource}/#", "sensors").resource, "sensors"); + } + + #[test] + fn magistrala_shaped_template_binds_tenant_and_resource() { + let result = matched("m/{tenant}/c/{resource}/#", "m/acme/c/telemetry/eu/1"); + assert_eq!(result.tenant.as_deref(), Some("acme")); + assert_eq!(result.resource, "telemetry"); + } + + #[test] + fn literal_segments_must_match() { + denied("m/{tenant}/c/{resource}/#", "x/acme/c/telemetry"); + denied("m/{tenant}/c/{resource}/#", "m/acme/d/telemetry"); + } + + #[test] + fn subtopic_binds_the_remainder() { + let result = matched("{resource}/{subtopic}", "telemetry/eu/rack1/temp"); + assert_eq!(result.resource, "telemetry"); + assert_eq!(result.subtopic.as_deref(), Some("eu/rack1/temp")); + } + + #[test] + fn empty_subtopic_remainder_is_none() { + assert_eq!(matched("{resource}/{subtopic}", "telemetry").subtopic, None); + } + + #[test] + fn single_wildcard_segment_is_discarded() { + let result = matched("{tenant}/+/{resource}", "acme/ignored/telemetry"); + assert_eq!(result.tenant.as_deref(), Some("acme")); + assert_eq!(result.resource, "telemetry"); + } + + // ── Broker wildcards ───────────────────────────────────────────────────── + + #[test] + fn wildcard_past_the_resource_is_allowed() { + assert_eq!( + matched("{resource}/#", "telemetry/+/temp").resource, + "telemetry" + ); + assert_eq!(matched("{resource}/#", "telemetry/#").resource, "telemetry"); + } + + #[test] + fn wildcard_on_the_resource_is_denied() { + denied("{resource}/#", "+/temp"); + denied("{resource}/#", "#"); + denied("m/{tenant}/c/{resource}/#", "m/acme/c/+/temp"); + } + + #[test] + fn wildcard_on_the_tenant_is_denied() { + denied("m/{tenant}/c/{resource}/#", "m/+/c/telemetry"); + denied("m/{tenant}/c/{resource}/#", "m/#"); + } + + #[test] + fn wildcard_on_a_literal_is_denied() { + denied("m/{tenant}/c/{resource}/#", "+/acme/c/telemetry"); + denied("m/{tenant}/c/{resource}/#", "m/acme/+/telemetry"); + } + + #[test] + fn hash_before_a_single_wildcard_still_denies_a_later_resource() { + // '#' at index 1 leaves the {resource} at index 2 unconstrained even + // though the template segment at index 1 is itself a wildcard. + denied("{tenant}/+/{resource}", "acme/#"); + } + + #[test] + fn non_terminal_hash_is_a_malformed_filter() { + denied("{resource}/#", "telemetry/#/temp"); + } + + // ── Length handling ────────────────────────────────────────────────────── + + #[test] + fn topic_longer_than_a_template_without_a_terminal_is_denied() { + denied("{tenant}/{resource}", "acme/telemetry/extra"); + } + + #[test] + fn topic_shorter_than_the_template_is_denied() { + denied("m/{tenant}/c/{resource}/#", "m/acme/c"); + denied("{tenant}/{resource}", "acme"); + } + + #[test] + fn multi_wildcard_matches_zero_remaining_segments() { + assert_eq!( + matched("{tenant}/{resource}/#", "acme/telemetry").resource, + "telemetry" + ); + } + + #[test] + fn empty_and_leading_slash_topics() { + denied("{resource}/#", ""); + assert_eq!( + matched("{resource}/#", "/telemetry/x").resource, + "telemetry" + ); + } + + #[test] + fn empty_bound_segment_is_denied() { + denied("{tenant}/{resource}", "acme/"); + } + + // ── Template validation ────────────────────────────────────────────────── + + #[test] + fn template_requires_exactly_one_resource() { + assert!(TopicTemplate::parse("{tenant}/#").is_err()); + assert!(TopicTemplate::parse("{resource}/{resource}").is_err()); + } + + #[test] + fn template_rejects_repeated_optional_placeholders() { + assert!(TopicTemplate::parse("{tenant}/{tenant}/{resource}").is_err()); + assert!(TopicTemplate::parse("{resource}/{subtopic}/{subtopic}").is_err()); + } + + #[test] + fn template_rejects_unknown_placeholders() { + assert!(TopicTemplate::parse("{domain}/{resource}").is_err()); + assert!(TopicTemplate::parse("{resource}/{}").is_err()); + } + + #[test] + fn template_rejects_wildcards_inside_literals() { + assert!(TopicTemplate::parse("m+/{resource}").is_err()); + } + + #[test] + fn template_rejects_segments_after_a_terminal() { + assert!(TopicTemplate::parse("{resource}/#/tail").is_err()); + assert!(TopicTemplate::parse("{resource}/{subtopic}/tail").is_err()); + } + + #[test] + fn template_rejects_empty_input() { + assert!(TopicTemplate::parse(" ").is_err()); + } + + // ── Template set ───────────────────────────────────────────────────────── + + #[test] + fn template_set_uses_the_first_binding_match() { + let set = TopicTemplateSet::parse_list(&[ + "m/{tenant}/c/{resource}/#".to_string(), + "{resource}/#".to_string(), + ]) + .expect("set should parse"); + + assert_eq!( + set.match_topic("m/acme/c/telemetry") + .unwrap() + .tenant + .as_deref(), + Some("acme") + ); + // Falls through to the second template rather than denying outright. + let plain = set.match_topic("telemetry/eu").unwrap(); + assert_eq!(plain.resource, "telemetry"); + assert_eq!(plain.tenant, None); + } + + #[test] + fn template_set_denies_when_no_template_binds() { + let set = TopicTemplateSet::parse_list(&["m/{tenant}/c/{resource}".to_string()]) + .expect("set should parse"); + assert!(set.match_topic("telemetry/eu").is_none()); + } + + #[test] + fn template_set_rejects_a_bad_member() { + assert!(TopicTemplateSet::parse_list(&["{resource}/#".into(), "{tenant}".into()]).is_err()); + } +} diff --git a/src/config.rs b/src/config.rs index e4a46ac2..4feefcb4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -75,6 +75,83 @@ pub struct Config { pub certs_root_ca_key_path: Option, pub certs_leaf_default_ttl_secs: u64, pub certs_leaf_max_ttl_secs: u64, + pub broker_auth: BrokerAuthConfig, +} + +/// How a topic segment addresses an object. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum BrokerTopicRef { + /// The bound segment is a tenant-scoped alias slug, resolved through the + /// same path as `AliasService.ResolveAlias`. + #[default] + Alias, + /// The bound segment is the object's UUID; no resolution step. + Uuid, +} + +impl BrokerTopicRef { + fn from_env_value(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "" | "alias" => Ok(Self::Alias), + "uuid" => Ok(Self::Uuid), + other => anyhow::bail!("ATOM_BROKER_TOPIC_REF must be alias or uuid, got {other}"), + } + } +} + +/// The broker auth callout — Atom implementing FluxMQ's `AuthService` so a +/// broker can call it with no adapter service in between. +/// +/// **Off by default, and deliberately so.** The callout has no bearer token to +/// check: a broker's gRPC client sends no `authorization` metadata, so the +/// endpoint authenticates its caller at the transport, via the gRPC listener's +/// mTLS client CA. Enabling it on a plaintext listener lets anything that can +/// reach the port authenticate and authorize as any principal. Enable it +/// together with `ATOM_GRPC_TLS_CLIENT_CA_PATH`, scoped to a CA that signs +/// brokers and nothing else. +#[derive(Debug, Clone)] +pub struct BrokerAuthConfig { + pub enabled: bool, + pub topic_templates: crate::broker_auth::TopicTemplateSet, + pub topic_ref: BrokerTopicRef, + /// Which credential kind a broker's username/password pair is checked + /// against. One kind, one lookup — the callout runs on the connect path and + /// trying both would double the cost of every rejected connection. + pub credential_kind: crate::models::enums::CredentialKind, +} + +/// First segment names the object, the rest is unconstrained — the near +/// universal MQTT convention, and the one that survives `+`/`#` in a filter. +pub const DEFAULT_BROKER_TOPIC_TEMPLATE: &str = "{resource}/#"; + +impl Default for BrokerAuthConfig { + fn default() -> Self { + Self { + enabled: false, + topic_templates: crate::broker_auth::TopicTemplateSet::parse_list(&[ + DEFAULT_BROKER_TOPIC_TEMPLATE.to_string(), + ]) + .expect("the built-in default template must parse"), + topic_ref: BrokerTopicRef::default(), + credential_kind: crate::models::enums::CredentialKind::Password, + } + } +} + +fn broker_credential_kind_from_env() -> Result { + use crate::models::enums::CredentialKind; + match std::env::var("ATOM_BROKER_CREDENTIAL_KIND") + .unwrap_or_default() + .trim() + .to_ascii_lowercase() + .as_str() + { + "" | "password" => Ok(CredentialKind::Password), + "shared_key" => Ok(CredentialKind::SharedKey), + other => { + anyhow::bail!("ATOM_BROKER_CREDENTIAL_KIND must be password or shared_key, got {other}") + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -540,6 +617,7 @@ impl Config { certs_root_ca_key_path: std::env::var("ATOM_CERTS_ROOT_CA_KEY_PATH").ok(), certs_leaf_default_ttl_secs: env_u64("ATOM_CERTS_LEAF_DEFAULT_TTL_SECS", 2_592_000), certs_leaf_max_ttl_secs: env_u64("ATOM_CERTS_LEAF_MAX_TTL_SECS", 2_592_000), + broker_auth: broker_auth_from_env()?, public_base_url, }) } @@ -606,6 +684,7 @@ impl Config { certs_root_ca_key_path: None, certs_leaf_default_ttl_secs: 2_592_000, certs_leaf_max_ttl_secs: 2_592_000, + broker_auth: BrokerAuthConfig::default(), } } } @@ -956,6 +1035,37 @@ fn grpc_tls_from_env() -> Result> { } } +fn broker_auth_from_env() -> Result { + let defaults = BrokerAuthConfig::default(); + let enabled = env_bool_default("ATOM_BROKER_AUTH_ENABLED", defaults.enabled); + if !enabled { + return Ok(defaults); + } + + let raw = nonempty_env("ATOM_BROKER_TOPIC_TEMPLATE") + .unwrap_or_else(|| DEFAULT_BROKER_TOPIC_TEMPLATE.to_string()); + // Templates are tried in order, so a deployment with more than one topic + // shape does not need a second Atom. + let templates: Vec = raw + .split(',') + .map(str::trim) + .filter(|template| !template.is_empty()) + .map(ToOwned::to_owned) + .collect(); + if templates.is_empty() { + anyhow::bail!("ATOM_BROKER_TOPIC_TEMPLATE must list at least one template"); + } + + Ok(BrokerAuthConfig { + enabled, + topic_templates: crate::broker_auth::TopicTemplateSet::parse_list(&templates)?, + topic_ref: BrokerTopicRef::from_env_value( + &std::env::var("ATOM_BROKER_TOPIC_REF").unwrap_or_default(), + )?, + credential_kind: broker_credential_kind_from_env()?, + }) +} + fn nonempty_env(name: &str) -> Option { std::env::var(name) .ok() diff --git a/src/grpc.rs b/src/grpc.rs index 54a87501..e7701f63 100644 --- a/src/grpc.rs +++ b/src/grpc.rs @@ -13,6 +13,7 @@ use crate::{ audit, auth::{authenticate_token, require_any_capability, scope_for_tenant, AuthContext, Scope}, authz::{access, engine, repo}, + broker_auth::{service::BrokerAuthServiceServer, BrokerAuth}, certs, identity::service as identity_service, models::{ @@ -565,6 +566,11 @@ pub async fn serve( health_reporter .set_serving::>() .await; + if state.config.broker_auth.enabled { + health_reporter + .set_serving::>() + .await; + } // The TLS config was already loaded and validated in `load_tls_config` // before this task was spawned (fail-fast at startup); here we only apply it. @@ -592,7 +598,7 @@ pub async fn serve( .set_grpc_status(GrpcRuntimeStatus::serving(addr.to_string())) .await; - let result = builder + let mut router = builder .add_service(health_service) .add_service(AuthzServiceServer::new(AtomAuthz { state: state.clone(), @@ -605,7 +611,41 @@ pub async fn serve( })) .add_service(AliasServiceServer::new(AtomAlias { state: state.clone(), - })) + })); + + // The broker callout is the one service here with no bearer token to check — + // a broker's callout client cannot send one — so its caller is authenticated + // at the transport instead. Mounting it without an mTLS client CA hands + // authentication and authorization for every principal to anything that can + // reach the port. + if state.config.broker_auth.enabled { + let client_ca = state + .config + .grpc_tls + .as_ref() + .and_then(|tls| tls.client_ca_path.as_deref()); + match client_ca { + Some(path) => tracing::info!( + client_ca = %path, + templates = ?state + .config + .broker_auth + .topic_templates + .iter() + .map(crate::broker_auth::TopicTemplate::as_str) + .collect::>(), + "broker auth callout enabled" + ), + None => tracing::warn!( + "broker auth callout enabled without ATOM_GRPC_TLS_CLIENT_CA_PATH; \ + any client that can reach this port can authenticate and authorize \ + as any principal" + ), + } + router = router.add_service(BrokerAuthServiceServer::new(BrokerAuth::new(state.clone()))); + } + + let result = router .serve_with_incoming_shutdown(incoming, crate::shutdown::shutdown_signal()) .await; diff --git a/src/lib.rs b/src/lib.rs index 5da29224..d681feb8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod api_endpoints; pub mod audit; pub mod auth; pub mod authz; +pub mod broker_auth; pub mod build_info; pub mod certs; pub mod config; diff --git a/tests/m29_broker_auth_callout.rs b/tests/m29_broker_auth_callout.rs new file mode 100644 index 00000000..225befd2 --- /dev/null +++ b/tests/m29_broker_auth_callout.rs @@ -0,0 +1,405 @@ +//! DB-gated tests for the broker auth callout — Atom serving FluxMQ's +//! `fluxmq.auth.v1.AuthService` directly, with no adapter service in between. +//! +//! Run with: +//! ```bash +//! DATABASE_URL=postgres://... cargo test --test m29_broker_auth_callout -- --ignored +//! ``` + +mod common; + +use atom::{ + authz::repo as authz_repo, + broker_auth::service::proto::{ + auth_service_client::AuthServiceClient, Action, AuthnReq, AuthzReq, + }, + config::{BrokerAuthConfig, BrokerTopicRef, Config}, + grpc, + identity::{repo as identity_repo, service as identity_service}, + keys::{self, ActiveKeys}, + models::{ + entity::CreateEntity, + enums::{EntityKind, SubjectKind}, + resource::CreateResource, + tenant::CreateTenant, + }, + state::AppState, + tenants::repo as tenant_repo, +}; +use serde_json::json; +use sqlx::PgPool; +use tokio::time::{sleep, Duration}; +use tonic::{transport::Channel, Request}; +use uuid::Uuid; + +const DEVICE_SECRET: &str = "broker-device-secret"; + +fn slug(prefix: &str) -> String { + let id = Uuid::new_v4().simple().to_string(); + format!("{prefix}-{}", &id[..12]) +} + +fn broker_config(template: &str, topic_ref: BrokerTopicRef) -> Config { + Config { + broker_auth: BrokerAuthConfig { + enabled: true, + topic_templates: atom::broker_auth::TopicTemplateSet::parse_list(&[ + template.to_string() + ]) + .expect("template parses"), + topic_ref, + ..BrokerAuthConfig::default() + }, + ..Config::for_tests() + } +} + +async fn active_keys(pool: &PgPool) -> ActiveKeys { + keys::rotate(pool, &Config::for_tests().signing_keys) + .await + .expect("rotate signing key") +} + +async fn make_tenant(pool: &PgPool) -> (Uuid, String) { + let alias = slug("dom"); + let tenant = tenant_repo::create_tenant( + pool, + CreateTenant { + id: None, + name: slug("tenant"), + alias: Some(alias.clone()), + tags: vec![], + attributes: json!({}), + }, + None, + ) + .await + .expect("create tenant"); + (tenant.id, alias) +} + +/// A device with a password credential, standing in for an MQTT client. +async fn make_device(pool: &PgPool, tenant_id: Option) -> (Uuid, String) { + let name = slug("dev"); + let device = identity_repo::create_entity( + pool, + CreateEntity { + id: None, + kind: Some(EntityKind::Device), + profile_id: None, + profile_version_id: None, + name: name.clone(), + alias: Some(slug("meter")), + tenant_id, + attributes: json!({}), + }, + ) + .await + .expect("create device"); + identity_service::create_password(pool, device.id, DEVICE_SECRET) + .await + .expect("create password"); + (device.id, name) +} + +/// A resource with an alias, standing in for an MQTT channel. +async fn make_channel(pool: &PgPool, tenant_id: Option) -> (Uuid, String) { + let alias = slug("chan"); + let resource = authz_repo::create_resource( + pool, + CreateResource { + id: None, + kind: "channel".to_string(), + name: Some(slug("channel")), + alias: Some(alias.clone()), + tenant_id, + owner_id: None, + attributes: json!({}), + }, + ) + .await + .expect("create resource"); + (resource.id, alias) +} + +/// Grant `action` on exactly one object, via a role assignment. +async fn grant(pool: &PgPool, subject: Uuid, tenant_id: Option, object: Uuid, action: &str) { + let role = authz_repo::create_role( + pool, + atom::models::role::CreateRole { + name: slug("broker-role"), + tenant_id, + description: None, + }, + ) + .await + .expect("create role"); + + let action_id: Uuid = sqlx::query_scalar("SELECT id FROM actions WHERE name = $1") + .bind(action) + .fetch_one(pool) + .await + .expect("seeded action"); + + let block: Uuid = sqlx::query_scalar( + "INSERT INTO permission_blocks (scope_mode, tenant_id, object_id, effect) + VALUES ('object', $1, $2, 'allow') RETURNING id", + ) + .bind(tenant_id) + .bind(object) + .fetch_one(pool) + .await + .expect("permission block"); + + sqlx::query( + "INSERT INTO permission_block_actions (permission_block_id, action_id) VALUES ($1, $2)", + ) + .bind(block) + .bind(action_id) + .execute(pool) + .await + .expect("block action"); + + authz_repo::replace_role_permission_block_links(pool, role.id, &[block]) + .await + .expect("link block"); + authz_repo::create_role_assignment( + pool, + atom::models::policy::CreateRoleAssignment { + tenant_id, + subject_kind: SubjectKind::Entity, + subject_id: subject, + role_id: role.id, + }, + ) + .await + .expect("assign role"); +} + +async fn serve(pool: &PgPool, cfg: Config) -> AuthServiceClient { + let keys = active_keys(pool).await; + let state = AppState::new(pool.clone(), cfg, keys, None); + let listener = grpc::bind_listener("127.0.0.1:0".parse().expect("addr")) + .await + .expect("bind grpc"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + let _ = grpc::serve(listener, state, None).await; + }); + + let endpoint = format!("http://{addr}"); + for _ in 0..40 { + if let Ok(channel) = Channel::from_shared(endpoint.clone()) + .expect("endpoint") + .connect() + .await + { + return AuthServiceClient::new(channel); + } + sleep(Duration::from_millis(25)).await; + } + panic!("gRPC server did not come up"); +} + +fn authn(username: &str, password: &str) -> Request { + Request::new(AuthnReq { + client_id: slug("mqtt"), + username: username.to_string(), + password: password.to_string(), + protocol: 0, + }) +} + +fn authz(external_id: &str, topic: &str, action: Action) -> Request { + Request::new(AuthzReq { + external_id: external_id.to_string(), + topic: topic.to_string(), + action: action as i32, + }) +} + +/// The zero-configuration path end to end: a device connects with its name and +/// password, and publishes to a topic whose first segment is a channel alias. +/// No tenant appears anywhere in the broker's requests — Atom derives it from +/// the authenticated subject. +#[tokio::test] +#[ignore] +async fn device_authenticates_and_publishes_with_no_tenant_in_the_topic() { + let pool = common::pool().await; + let (tenant_id, _) = make_tenant(&pool).await; + let (device_id, device_name) = make_device(&pool, Some(tenant_id)).await; + let (channel_id, channel_alias) = make_channel(&pool, Some(tenant_id)).await; + grant(&pool, device_id, Some(tenant_id), channel_id, "publish").await; + + let mut client = serve(&pool, broker_config("{resource}/#", BrokerTopicRef::Alias)).await; + + let authenticated = client + .authenticate(authn(&device_name, DEVICE_SECRET)) + .await + .expect("authenticate rpc") + .into_inner(); + assert!(authenticated.authenticated); + assert_eq!(authenticated.id, device_id.to_string()); + + let allowed = client + .authorize(authz( + &authenticated.id, + &format!("{channel_alias}/eu/temp"), + Action::Publish, + )) + .await + .expect("authorize rpc") + .into_inner(); + assert!( + allowed.authorized, + "publish should be allowed: {}", + allowed.reason + ); + + // Only `publish` was granted. + let denied = client + .authorize(authz(&authenticated.id, &channel_alias, Action::Subscribe)) + .await + .expect("authorize rpc") + .into_inner(); + assert!(!denied.authorized); +} + +/// A `{tenant}` segment scopes alias resolution, so a topic can address a +/// channel without Atom consulting the subject's own tenant. +#[tokio::test] +#[ignore] +async fn tenant_segment_scopes_alias_resolution() { + let pool = common::pool().await; + let (tenant_id, tenant_alias) = make_tenant(&pool).await; + let (device_id, _) = make_device(&pool, Some(tenant_id)).await; + let (channel_id, channel_alias) = make_channel(&pool, Some(tenant_id)).await; + grant(&pool, device_id, Some(tenant_id), channel_id, "publish").await; + + let mut client = serve( + &pool, + broker_config("m/{tenant}/c/{resource}/#", BrokerTopicRef::Alias), + ) + .await; + + let allowed = client + .authorize(authz( + &device_id.to_string(), + &format!("m/{tenant_alias}/c/{channel_alias}/eu"), + Action::Publish, + )) + .await + .expect("authorize rpc") + .into_inner(); + assert!(allowed.authorized, "{}", allowed.reason); +} + +/// UUID topics skip alias resolution entirely. +#[tokio::test] +#[ignore] +async fn uuid_topic_ref_addresses_the_object_directly() { + let pool = common::pool().await; + let (tenant_id, _) = make_tenant(&pool).await; + let (device_id, _) = make_device(&pool, Some(tenant_id)).await; + let (channel_id, _) = make_channel(&pool, Some(tenant_id)).await; + grant(&pool, device_id, Some(tenant_id), channel_id, "subscribe").await; + + let mut client = serve(&pool, broker_config("{resource}/#", BrokerTopicRef::Uuid)).await; + + let allowed = client + .authorize(authz( + &device_id.to_string(), + &format!("{channel_id}/eu/temp"), + Action::Subscribe, + )) + .await + .expect("authorize rpc") + .into_inner(); + assert!(allowed.authorized, "{}", allowed.reason); +} + +/// Every rejection must arrive as a successful RPC carrying a false verdict. +/// A gRPC error would trip the broker's circuit breaker, which rejects *all* +/// client connections — one bad device must not be able to cause that. +#[tokio::test] +#[ignore] +async fn every_rejection_is_a_verdict_not_an_rpc_error() { + let pool = common::pool().await; + let (tenant_id, _) = make_tenant(&pool).await; + let (device_id, device_name) = make_device(&pool, Some(tenant_id)).await; + let (channel_id, channel_alias) = make_channel(&pool, Some(tenant_id)).await; + grant(&pool, device_id, Some(tenant_id), channel_id, "publish").await; + + let mut client = serve(&pool, broker_config("{resource}/#", BrokerTopicRef::Alias)).await; + + let wrong_password = client + .authenticate(authn(&device_name, "not-the-secret")) + .await + .expect("wrong password must not be an rpc error") + .into_inner(); + assert!(!wrong_password.authenticated); + assert!(wrong_password.id.is_empty()); + + let unknown_user = client + .authenticate(authn(&slug("ghost"), DEVICE_SECRET)) + .await + .expect("unknown identity must not be an rpc error") + .into_inner(); + assert!(!unknown_user.authenticated); + + let empty = client + .authenticate(authn("", "")) + .await + .expect("empty credentials must not be an rpc error") + .into_inner(); + assert!(!empty.authenticated); + + for (topic, why) in [ + (format!("{channel_alias}/x"), "granted, for contrast"), + (slug("nosuchchannel"), "unknown object"), + ("+/temp".to_string(), "wildcard on the object segment"), + ("#".to_string(), "wildcard spanning everything"), + ] { + let response = client + .authorize(authz(&device_id.to_string(), &topic, Action::Publish)) + .await + .unwrap_or_else(|err| panic!("{why} must not be an rpc error: {err}")) + .into_inner(); + if why == "granted, for contrast" { + assert!(response.authorized, "{why}: {}", response.reason); + } else { + assert!(!response.authorized, "{why} should be denied"); + } + } + + // A client the broker never authenticated arrives with a protocol-level id. + let unauthenticated = client + .authorize(authz("mqtt-client-7", &channel_alias, Action::Publish)) + .await + .expect("non-UUID subject must not be an rpc error") + .into_inner(); + assert!(!unauthenticated.authorized); + + // `Action::None` is the proto's unset value. + let no_action = client + .authorize(authz(&device_id.to_string(), &channel_alias, Action::None)) + .await + .expect("unset action must not be an rpc error") + .into_inner(); + assert!(!no_action.authorized); +} + +/// The callout is off unless explicitly enabled, because it authenticates its +/// caller at the transport rather than with a bearer token. +#[tokio::test] +#[ignore] +async fn callout_is_not_mounted_unless_enabled() { + let pool = common::pool().await; + let mut client = serve(&pool, Config::for_tests()).await; + + let status = client + .authenticate(authn("someone", "something")) + .await + .expect_err("service must not be mounted when disabled"); + assert_eq!(status.code(), tonic::Code::Unimplemented); +} From 73f4e63c872128405ea19f830edfc6ed5ada1a5a Mon Sep 17 00:00:00 2001 From: dusan Date: Thu, 6 Aug 2026 20:16:42 +0200 Subject: [PATCH 2/6] Admit broker operational topics without the PDP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A broker carries topics that address no object — a health probe such as hc/ names nothing Atom can resolve — so no policy could describe them and every request for one was denied. ATOM_BROKER_TOPIC_ALLOW lists topics authorized without consulting the PDP. This is the only authorization bypass in the callout, so it defaults to empty and its patterns are ordinary MQTT filters: an operator writes the narrowest shape that covers the operational topic rather than reaching for a prefix. A broker '#' covers its position and everything below it, so only a pattern that is itself '#' there is broad enough to admit it. Letting '+' match it would silently widen the bypass past what the operator wrote — hc/+ would admit a subscription to the whole hc subtree. --- AGENTS.md | 11 +++ src/broker_auth/mod.rs | 2 +- src/broker_auth/service.rs | 16 +++ src/broker_auth/topic.rs | 165 +++++++++++++++++++++++++++++++ src/config.rs | 23 +++++ tests/m29_broker_auth_callout.rs | 55 +++++++++++ 6 files changed, 271 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 23477160..be79b122 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -243,6 +243,17 @@ Config (`ATOM_BROKER_*`), all optional: | `ATOM_BROKER_TOPIC_TEMPLATE` | `{resource}/#` | comma-separated templates, tried in order | | `ATOM_BROKER_TOPIC_REF` | `alias` | `alias` or `uuid` — how a bound segment names an object | | `ATOM_BROKER_CREDENTIAL_KIND` | `password` | `password` or `shared_key` | +| `ATOM_BROKER_TOPIC_ALLOW` | *(empty)* | comma-separated MQTT filters authorized **without consulting the PDP** | + +`ATOM_BROKER_TOPIC_ALLOW` is the only authorization bypass in the callout. It +exists because brokers carry operational topics that address no object — a +health probe such as `hc/` names nothing Atom can resolve, so no policy +could describe it and every request for it would be denied. Patterns are +ordinary MQTT filters (`+` one segment, `#` the remainder); use the narrowest +one that covers the topic, since `#` alone grants the broker everything. The +broker's topic is matched literally and a broker `#` is only admitted by a +pattern that is itself `#` at that position — otherwise `hc/+` would quietly +admit a subscription to the whole `hc` subtree. **Off by default for a security reason, not a rollout one.** It is the only gRPC service here with no bearer token to check — a broker's callout client cannot diff --git a/src/broker_auth/mod.rs b/src/broker_auth/mod.rs index dc791e17..2d4a10a8 100644 --- a/src/broker_auth/mod.rs +++ b/src/broker_auth/mod.rs @@ -20,4 +20,4 @@ pub mod service; pub mod topic; pub use service::BrokerAuth; -pub use topic::{TopicMatch, TopicTemplate, TopicTemplateSet}; +pub use topic::{TopicAllowList, TopicMatch, TopicTemplate, TopicTemplateSet}; diff --git a/src/broker_auth/service.rs b/src/broker_auth/service.rs index 82fc8615..211046c1 100644 --- a/src/broker_auth/service.rs +++ b/src/broker_auth/service.rs @@ -158,6 +158,22 @@ impl AuthService for BrokerAuth { return Ok(Response::new(authz_denied("unsupported action"))); }; + // Operational topics are admitted before anything else, because they + // address no object: there is nothing for the PDP to decide on, and no + // policy an operator could write that would allow them. Empty unless a + // deployment configures it. + if self.state.config.broker_auth.topic_allow.allows(&req.topic) { + tracing::debug!( + external_id = %req.external_id, topic = %req.topic, + "broker authorize: allow (operational topic)" + ); + return Ok(Response::new(AuthzRes { + authorized: true, + reason_code: REASON_SUCCESS, + reason: String::new(), + })); + } + // `external_id` is whatever Authenticate returned. If the broker did not // authenticate this client it passes the protocol-level client id, which // is not an Atom subject — deny rather than guess. diff --git a/src/broker_auth/topic.rs b/src/broker_auth/topic.rs index 9869c11f..f8f2987c 100644 --- a/src/broker_auth/topic.rs +++ b/src/broker_auth/topic.rs @@ -286,6 +286,93 @@ impl TopicTemplateSet { } } +/// Topics allowed without consulting the PDP. +/// +/// **This is an authorization bypass**, and the only one in the callout. It +/// exists because brokers carry operational topics that address no object at +/// all — a health probe such as `hc/` names nothing Atom can resolve, so +/// there is no policy that could describe it and every request for it would be +/// denied. Defaults to empty; a deployment that does not need it never gets one. +/// +/// Patterns use ordinary MQTT filter syntax — `+` for one segment, `#` for the +/// remainder — so `hc/+` admits a per-tenant health topic without also admitting +/// `hc/a/b`. Prefer the narrowest pattern that covers the operational topic; +/// `#` alone would hand the broker unconditional access to everything. +/// +/// The broker's topic is matched literally, so a subscription to `hc/#` is +/// admitted only by a pattern that itself covers `#` at that position. +#[derive(Debug, Clone, Default)] +pub struct TopicAllowList { + filters: Vec>, +} + +impl TopicAllowList { + pub fn parse_list(patterns: &[String]) -> Result { + let mut filters = Vec::new(); + for pattern in patterns { + let raw = pattern.trim(); + let fail = |reason: &str| TemplateParseError { + template: raw.to_string(), + reason: reason.to_string(), + }; + if raw.is_empty() { + return Err(fail("allow pattern must not be empty")); + } + let segments: Vec = raw.split('/').map(ToOwned::to_owned).collect(); + if let Some(index) = segments.iter().position(|segment| segment == "#") { + if index != segments.len() - 1 { + return Err(fail("'#' must be the last segment")); + } + } + filters.push(segments); + } + Ok(Self { filters }) + } + + pub fn is_empty(&self) -> bool { + self.filters.is_empty() + } + + pub fn allows(&self, topic: &str) -> bool { + let topic = topic.strip_prefix('/').unwrap_or(topic); + if topic.is_empty() { + return false; + } + let tokens: Vec<&str> = topic.split('/').collect(); + self.filters + .iter() + .any(|filter| filter_matches(filter, &tokens)) + } +} + +fn filter_matches(filter: &[String], tokens: &[&str]) -> bool { + let mut index = 0; + while index < filter.len() { + // A broker '#' covers this position *and everything below it*, so only + // a pattern that is itself '#' here is broad enough to admit it. Letting + // '+' match it would widen the bypass past what the operator wrote — + // `hc/+` would admit a subscription to the whole `hc` subtree. + if filter[index] != "#" && tokens.get(index) == Some(&"#") { + return false; + } + match filter[index].as_str() { + "#" => return true, + "+" => { + if index >= tokens.len() { + return false; + } + } + literal => { + if tokens.get(index) != Some(&literal) { + return false; + } + } + } + index += 1; + } + index == tokens.len() +} + #[cfg(test)] mod tests { use super::*; @@ -499,4 +586,82 @@ mod tests { fn template_set_rejects_a_bad_member() { assert!(TopicTemplateSet::parse_list(&["{resource}/#".into(), "{tenant}".into()]).is_err()); } + + // ── Allow list ─────────────────────────────────────────────────────────── + + fn allow(patterns: &[&str]) -> TopicAllowList { + TopicAllowList::parse_list(&patterns.iter().map(ToString::to_string).collect::>()) + .expect("patterns should parse") + } + + #[test] + fn the_allow_list_is_empty_by_default() { + let list = TopicAllowList::default(); + assert!(list.is_empty()); + assert!(!list.allows("hc/acme")); + } + + #[test] + fn a_single_segment_wildcard_admits_a_per_tenant_health_topic() { + let list = allow(&["hc/+"]); + assert!(list.allows("hc/acme")); + assert!(list.allows("hc/00000000-0000-0000-0000-000000000001")); + } + + #[test] + fn a_single_segment_wildcard_does_not_admit_deeper_topics() { + let list = allow(&["hc/+"]); + assert!(!list.allows("hc/acme/extra")); + assert!(!list.allows("hc")); + } + + #[test] + fn an_unrelated_topic_is_never_admitted() { + let list = allow(&["hc/+"]); + assert!(!list.allows("m/acme/c/telemetry")); + assert!(!list.allows("telemetry")); + assert!(!list.allows("")); + } + + #[test] + fn a_multi_segment_wildcard_admits_the_whole_subtree() { + let list = allow(&["$sys/#"]); + assert!(list.allows("$sys")); + assert!(list.allows("$sys/broker/uptime")); + assert!(!list.allows("sys/broker")); + } + + #[test] + fn an_exact_pattern_admits_only_itself() { + let list = allow(&["hc"]); + assert!(list.allows("hc")); + assert!(!list.allows("hc/acme")); + } + + #[test] + fn any_pattern_in_the_list_may_admit() { + let list = allow(&["hc/+", "$sys/#"]); + assert!(list.allows("hc/acme")); + assert!(list.allows("$sys/uptime")); + assert!(!list.allows("m/acme/c/telemetry")); + } + + #[test] + fn a_leading_slash_is_ignored_as_it_is_for_templates() { + assert!(allow(&["hc/+"]).allows("/hc/acme")); + } + + #[test] + fn broker_wildcards_are_matched_literally() { + // A subscription to `hc/#` spans more than `hc/+` describes, so only a + // pattern that itself covers the position admits it. + assert!(!allow(&["hc/+"]).allows("hc/#")); + assert!(allow(&["hc/#"]).allows("hc/#")); + } + + #[test] + fn allow_patterns_reject_a_non_terminal_hash_and_empty_input() { + assert!(TopicAllowList::parse_list(&["hc/#/tail".to_string()]).is_err()); + assert!(TopicAllowList::parse_list(&[" ".to_string()]).is_err()); + } } diff --git a/src/config.rs b/src/config.rs index 4feefcb4..c8a90023 100644 --- a/src/config.rs +++ b/src/config.rs @@ -118,6 +118,10 @@ pub struct BrokerAuthConfig { /// against. One kind, one lookup — the callout runs on the connect path and /// trying both would double the cost of every rejected connection. pub credential_kind: crate::models::enums::CredentialKind, + /// Topics authorized without consulting the PDP. Empty by default; see + /// [`crate::broker_auth::TopicAllowList`] for why it exists and how narrow + /// a pattern should be. + pub topic_allow: crate::broker_auth::TopicAllowList, } /// First segment names the object, the rest is unconstrained — the near @@ -134,10 +138,26 @@ impl Default for BrokerAuthConfig { .expect("the built-in default template must parse"), topic_ref: BrokerTopicRef::default(), credential_kind: crate::models::enums::CredentialKind::Password, + topic_allow: crate::broker_auth::TopicAllowList::default(), } } } +/// Split a comma-separated env var, dropping blanks. Absent or blank yields an +/// empty list. +fn comma_list(name: &str) -> Vec { + nonempty_env(name) + .map(|value| { + value + .split(',') + .map(str::trim) + .filter(|entry| !entry.is_empty()) + .map(ToOwned::to_owned) + .collect() + }) + .unwrap_or_default() +} + fn broker_credential_kind_from_env() -> Result { use crate::models::enums::CredentialKind; match std::env::var("ATOM_BROKER_CREDENTIAL_KIND") @@ -1063,6 +1083,9 @@ fn broker_auth_from_env() -> Result { &std::env::var("ATOM_BROKER_TOPIC_REF").unwrap_or_default(), )?, credential_kind: broker_credential_kind_from_env()?, + topic_allow: crate::broker_auth::TopicAllowList::parse_list(&comma_list( + "ATOM_BROKER_TOPIC_ALLOW", + ))?, }) } diff --git a/tests/m29_broker_auth_callout.rs b/tests/m29_broker_auth_callout.rs index 225befd2..3280ed13 100644 --- a/tests/m29_broker_auth_callout.rs +++ b/tests/m29_broker_auth_callout.rs @@ -389,6 +389,61 @@ async fn every_rejection_is_a_verdict_not_an_rpc_error() { assert!(!no_action.authorized); } +/// Operational topics address no object, so no policy could describe them and +/// every request for one would otherwise be denied. The allow list is the only +/// path in the callout that skips the PDP, so its edges matter: it must admit +/// exactly the configured shape and nothing broader. +#[tokio::test] +#[ignore] +async fn the_allow_list_admits_operational_topics_and_nothing_wider() { + let pool = common::pool().await; + let (tenant_id, _) = make_tenant(&pool).await; + let (device_id, _) = make_device(&pool, Some(tenant_id)).await; + let (_, channel_alias) = make_channel(&pool, Some(tenant_id)).await; + + let mut cfg = broker_config("{resource}/#", BrokerTopicRef::Alias); + cfg.broker_auth.topic_allow = + atom::broker_auth::TopicAllowList::parse_list(&["hc/+".to_string()]) + .expect("allow list parses"); + let mut client = serve(&pool, cfg).await; + + let health = client + .authorize(authz(&device_id.to_string(), "hc/acme", Action::Publish)) + .await + .expect("authorize rpc") + .into_inner(); + assert!(health.authorized, "health topic should be admitted"); + + // The device holds no grant on this channel; the allow list must not have + // turned into a general bypass. + let ungranted = client + .authorize(authz( + &device_id.to_string(), + &channel_alias, + Action::Publish, + )) + .await + .expect("authorize rpc") + .into_inner(); + assert!( + !ungranted.authorized, + "allow list leaked into ordinary topics" + ); + + for (topic, why) in [ + ("hc/acme/extra", "deeper than the single-segment pattern"), + ("hc", "shorter than the pattern"), + ("hc/#", "spans the whole subtree, wider than `hc/+`"), + ] { + let response = client + .authorize(authz(&device_id.to_string(), topic, Action::Publish)) + .await + .expect("authorize rpc") + .into_inner(); + assert!(!response.authorized, "{topic} admitted, but it is {why}"); + } +} + /// The callout is off unless explicitly enabled, because it authenticates its /// caller at the transport rather than with a bearer token. #[tokio::test] From 4a087adc44284d918d51243da2d305760a564adb Mon Sep 17 00:00:00 2001 From: dusan Date: Fri, 7 Aug 2026 11:10:44 +0200 Subject: [PATCH 3/6] Serve the broker callout under a vendor-neutral package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wire path a broker dials is derived from the proto package, so serving `fluxmq.auth.v1` put one implementation's name in the public surface of every peer that speaks the contract — including Atom's. Nothing in the messages is FluxMQ-specific; the same shape covers MQTT, AMQP, CoAP and HTTP, and other brokers and providers implement it. Renaming costs nothing structurally. The Go import path upstream is derived from the file's location rather than the package, so consumers keep the same imports and symbols, and only the dialled path changes. It is still a breaking wire change: a broker dialling the new path against a service serving the old one gets UNIMPLEMENTED. Atom, the brokers, and any adapter service have to move together. --- proto/broker/v1/auth.proto | 21 ++++++++------------- src/broker_auth/mod.rs | 2 +- src/broker_auth/service.rs | 4 ++-- 3 files changed, 11 insertions(+), 16 deletions(-) diff --git a/proto/broker/v1/auth.proto b/proto/broker/v1/auth.proto index 8f62a903..75963f1d 100644 --- a/proto/broker/v1/auth.proto +++ b/proto/broker/v1/auth.proto @@ -1,21 +1,16 @@ // Copyright (c) Abstract Machines // SPDX-License-Identifier: Apache-2.0 -// Vendored verbatim from FluxMQ `proto/auth/v1/auth.proto`. Atom implements -// `AuthService` so a broker can call it directly, with no adapter service in -// between. Check for drift with: -// -// diff proto/broker/v1/auth.proto \ -// $FLUXMQ/proto/auth/v1/auth.proto -// -// The `package` line is part of the wire contract — the gRPC path a broker -// dials is `/fluxmq.auth.v1.AuthService/Authorize`, derived from the proto -// package plus service name. Renaming it here silently stops matching. -// `HookService` is vendored for completeness but is not implemented. - syntax = "proto3"; -package fluxmq.auth.v1; +// The package is vendor-neutral on purpose. These services are a generic +// broker-callout contract — the messages carry no FluxMQ concept, and other +// brokers and providers implement them — but the proto package is what a caller +// dials (`/broker.auth.v1.AuthService/Authorize`), so naming it after one +// implementation would put that implementation's name in every peer's public +// wire surface. The Go import path is derived from this file's location, not +// from the package, so it is unaffected. +package broker.auth.v1; // AuthService is a callout service for external authentication and // authorization. Broker implementations call this service during connection diff --git a/src/broker_auth/mod.rs b/src/broker_auth/mod.rs index 2d4a10a8..1d4580af 100644 --- a/src/broker_auth/mod.rs +++ b/src/broker_auth/mod.rs @@ -6,7 +6,7 @@ //! service in between. //! //! The contract is deliberately the broker's, not Atom's: the wire path a -//! broker dials is `/fluxmq.auth.v1.AuthService/...`, derived from the vendored +//! broker dials is `/broker.auth.v1.AuthService/...`, derived from the vendored //! proto's package. Everything Atom needs beyond that — how a topic names an //! object — is configuration, so Atom never learns a particular deployment's //! topic vocabulary. See [`topic`] for the grammar. diff --git a/src/broker_auth/service.rs b/src/broker_auth/service.rs index 211046c1..d1014d22 100644 --- a/src/broker_auth/service.rs +++ b/src/broker_auth/service.rs @@ -1,4 +1,4 @@ -//! The `fluxmq.auth.v1.AuthService` implementation. +//! The `broker.auth.v1.AuthService` implementation. //! //! ## Denials are answers, not errors //! @@ -33,7 +33,7 @@ use super::topic::TopicMatch; // Generated from the vendored proto/broker/v1/auth.proto. The module path is // the proto package, which is also the gRPC wire path a broker dials. pub mod proto { - tonic::include_proto!("fluxmq.auth.v1"); + tonic::include_proto!("broker.auth.v1"); } pub use proto::auth_service_server::AuthServiceServer as BrokerAuthServiceServer; From 5a9848e9a142fb57e755efef62dafa549634739f Mon Sep 17 00:00:00 2001 From: dusan Date: Fri, 7 Aug 2026 11:11:36 +0200 Subject: [PATCH 4/6] Detect drift in the vendored broker contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Atom implements a proto it does not own, and nothing rebuilds the copy. An upstream change was therefore discovered at runtime — as an UNIMPLEMENTED from a renamed service, or worse, as a field that still decodes but no longer means what Atom thinks it means. scripts/check-vendored-proto.sh diffs the vendored file against the ref pinned in proto/broker/v1/REF, and CI runs it. For that diff to stay trustworthy the copy has to be byte-identical, so Atom's own notes moved out of the proto and into VENDOR.md beside it: a check that has to forgive expected differences stops catching the one that matters. Vendoring a second proto into the buf module had also broken two things that nothing here runs, so neither had surfaced: - buf lint failed on twenty-odd violations in the vendored file. Its style is upstream's, and Atom cannot fix it without breaking the byte-for-byte match, so it is excluded from lint and breaking. - buf generate silently replaced apidocs/grpc-reference.md with the broker contract, dropping Atom's own gRPC surface from the docs. protoc-gen-doc writes one file per invocation, so a second package does not extend that file. Generation is now scoped to Atom's protos. make proto regenerates both outputs. The asymmetry is worth knowing: the Rust bindings are not checked in — build.rs runs tonic-build into OUT_DIR on every compile — but grpc-reference.md is, and goes stale silently. --- .github/workflows/rust.yml | 5 +++ AGENTS.md | 33 ++++++++++++++++--- Makefile | 35 +++++++++++++++++++- buf.gen.yaml | 10 ++++++ buf.yaml | 12 ++++++- proto/broker/v1/REF | 1 + proto/broker/v1/VENDOR.md | 49 ++++++++++++++++++++++++++++ scripts/check-vendored-proto.sh | 57 +++++++++++++++++++++++++++++++++ 8 files changed, 196 insertions(+), 6 deletions(-) create mode 100644 proto/broker/v1/REF create mode 100644 proto/broker/v1/VENDOR.md create mode 100755 scripts/check-vendored-proto.sh diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index cc89e8c7..8362fb15 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -48,6 +48,11 @@ jobs: with: repo-token: ${{ secrets.GITHUB_TOKEN }} + # Atom implements a proto it does not own, and nothing rebuilds the + # vendored copy. Without this, an upstream change surfaces at runtime. + - name: Vendored proto matches upstream + run: scripts/check-vendored-proto.sh + - name: cargo fmt run: cargo fmt --check diff --git a/AGENTS.md b/AGENTS.md index be79b122..4e55d83a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,7 @@ src/ keys.rs — ES256 signing keys (primary/standby/retired), encryption at rest grpc.rs — Tonic services: AuthService, AuthzService.Check, CertificateService broker_auth/ — the broker auth callout: Atom serving FluxMQ's - │ `fluxmq.auth.v1.AuthService` directly (off by default) + │ `broker.auth.v1.AuthService` directly (off by default) │ topic.rs — the configurable topic→object grammar │ service.rs — Authenticate/Authorize over the existing credential + PDP paths graphql/ — schema + per-domain resolvers (the live admin/API surface) @@ -204,6 +204,14 @@ cargo test -- --include-ignored # Lint cargo clippy -- -D warnings cargo fmt --check + +# Protobuf. The Rust bindings are NOT checked in — build.rs runs tonic-build on +# every compile into cargo's OUT_DIR, so editing a .proto and rebuilding is +# enough for code. `make proto` also regenerates apidocs/grpc-reference.md, +# which IS checked in and goes stale silently without it. +make proto +make proto-lint # protos Atom owns +make proto-check # vendored broker contract vs upstream ``` Environment variables: copy `.env.example` to `.env`. Required: `DATABASE_URL`. Signing uses ES256 keys bootstrapped/loaded at startup — there is no `JWT_SECRET`. `ATOM_KEY_ENCRYPTION_KEY` is the single root AES-256-GCM key encrypting all recoverable secrets at rest (signing private keys and retrievable credential secrets such as shared keys); it is required to create shared keys. @@ -231,9 +239,26 @@ A message broker delegates connect-time credential checks and per-topic access control to an external gRPC service. Atom implements that contract itself (`src/broker_auth/`), so a broker can be pointed straight at Atom with **no adapter service in between**. The proto is vendored verbatim from FluxMQ at -`proto/broker/v1/auth.proto`; its `package fluxmq.auth.v1` line is part of the -wire contract (the dialled path is `/fluxmq.auth.v1.AuthService/Authorize`) and -must not be renamed. Check for drift with a `diff` against the FluxMQ checkout. +`proto/broker/v1/auth.proto` as a **byte-identical** copy, so drift is a plain +`diff` — `scripts/check-vendored-proto.sh`, which CI runs against the ref pinned +in `proto/broker/v1/REF`. Atom's notes live beside it in `VENDOR.md`, never in +the proto itself: a check that has to forgive expected differences stops +catching the one that matters. + +Because Atom does not own that file, it is excluded from `buf.yaml`'s lint and +breaking rules (its style is upstream's, and editing it would break the +byte-for-byte match) and from `buf.gen.yaml`'s inputs — `protoc-gen-doc` writes +one file per invocation, so including a second package does not extend +`apidocs/grpc-reference.md`, it **replaces** it and Atom's own gRPC surface +vanishes from the docs. + +The `package broker.auth.v1` line **is** the contract — the path a broker dials +is derived from it. It is vendor-neutral on purpose: the messages carry no +FluxMQ concept, so naming the package after one implementation would put that +name in every peer's public wire surface. Changing it is a breaking wire change: +a broker dialling the new path against a service still serving the old one gets +`UNIMPLEMENTED`, so Atom, the broker, and any adapter service must be deployed +together. Config (`ATOM_BROKER_*`), all optional: diff --git a/Makefile b/Makefile index 0ece14a8..ab7a5e01 100644 --- a/Makefile +++ b/Makefile @@ -30,7 +30,7 @@ COMPOSE_ENV = ATOM_IMAGE="$(ATOM_IMAGE)" ATOM_UI_IMAGE="$(ATOM_UI_IMAGE)" DEV_HTTP_PORT ?= 8090 DEV_UI_PORT ?= 3000 -.PHONY: help db dev build latest release release-check atom-build docker_atom_dev ui-build up down logs restart docker-build docker-build-release +.PHONY: help db dev build latest release release-check atom-build docker_atom_dev ui-build up down logs restart docker-build docker-build-release proto proto-lint proto-check help: @echo "First run: cp .env.example .env" @@ -46,6 +46,9 @@ help: @echo " make db Start only Postgres (for host 'cargo run')" @echo " make dev Postgres (Docker) + host cargo run (:$(DEV_HTTP_PORT)) + host UI (:$(DEV_UI_PORT)); runs alongside 'make up'" @echo " make restart Restart the Compose stack (no rebuild; use 'make build' first)" + @echo " make proto Regenerate protobuf outputs (gRPC reference docs + Rust bindings)" + @echo " make proto-lint Lint the protos Atom owns" + @echo " make proto-check Verify the vendored broker contract still matches upstream" @echo " make logs Follow Atom + Atom UI logs" @echo " make down Stop the local Compose stack" @echo " make docker-build Build the raw Atom Docker image for BUILD_TARGET" @@ -170,3 +173,33 @@ docker-build: docker-build-release: $(MAKE) docker-build BUILD_TARGET=release IMAGE_TAG=release + +# ─── Protobuf ───────────────────────────────────────────────────────────────── +# +# Atom has two protobuf outputs, and only one of them is a file in the repo: +# +# apidocs/grpc-reference.md — checked in, produced by `buf generate` +# the Rust service bindings — NOT checked in; build.rs runs tonic-build on +# every compile and writes into cargo's OUT_DIR +# +# So this target regenerates the docs and then rebuilds, which is what refreshes +# the bindings. Editing a .proto and running `cargo build` is enough on its own — +# tonic-build emits `cargo:rerun-if-changed` for each proto — but the docs are +# generated by buf and will silently go stale without this. +proto: + @command -v buf >/dev/null || { \ + echo "buf not found — install from https://buf.build/docs/installation"; exit 1; } + @command -v protoc-gen-doc >/dev/null || { \ + echo "protoc-gen-doc not found — go install github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc@latest"; exit 1; } + buf generate + cargo build + +# The vendored broker contract is excluded in buf.yaml: Atom does not own its +# style, and editing it would break the byte-for-byte match `proto-check` needs. +proto-lint: + buf lint + +# Upstream owns proto/broker/v1/auth.proto. Nothing rebuilds it, so without this +# an upstream change surfaces at runtime. CI runs the same script. +proto-check: + scripts/check-vendored-proto.sh diff --git a/buf.gen.yaml b/buf.gen.yaml index 3ed6f082..1b0b80fe 100644 --- a/buf.gen.yaml +++ b/buf.gen.yaml @@ -1,5 +1,15 @@ version: v2 +# Only the protos Atom owns. proto/broker/v1/auth.proto is vendored verbatim +# from FluxMQ (see its VENDOR.md) and is deliberately excluded: protoc-gen-doc +# writes one file per invocation, so a second package here does not extend +# grpc-reference.md — it silently replaces it, and Atom's own gRPC surface +# disappears from the docs. The vendored contract is documented upstream. +inputs: + - directory: proto + exclude_paths: + - proto/broker + plugins: # Generates apidocs/grpc-reference.md from the proto. # Requires protoc-gen-doc: go install github.com/pseudomuto/protoc-gen-doc/cmd/protoc-gen-doc@latest diff --git a/buf.yaml b/buf.yaml index 53175feb..a4b07d66 100644 --- a/buf.yaml +++ b/buf.yaml @@ -8,8 +8,18 @@ lint: - STANDARD except: - RPC_REQUEST_RESPONSE_UNIQUE + # proto/broker/v1/auth.proto is vendored verbatim from FluxMQ (see its + # VENDOR.md). Atom does not own its style and cannot fix it without breaking + # the byte-for-byte match the drift check depends on, so linting it would + # only produce failures no one here can act on. + ignore: + - proto/broker/v1 breaking: use: - FILE - + # Same file, different reason: a change here is upstream's, and it is caught + # by scripts/check-vendored-proto.sh, which reports it as drift with the + # context needed to judge it. + ignore: + - proto/broker/v1 diff --git a/proto/broker/v1/REF b/proto/broker/v1/REF new file mode 100644 index 00000000..ba2906d0 --- /dev/null +++ b/proto/broker/v1/REF @@ -0,0 +1 @@ +main diff --git a/proto/broker/v1/VENDOR.md b/proto/broker/v1/VENDOR.md new file mode 100644 index 00000000..d7b985a5 --- /dev/null +++ b/proto/broker/v1/VENDOR.md @@ -0,0 +1,49 @@ +# Vendored broker-callout contract + +`auth.proto` is a **byte-identical copy** of FluxMQ's +`proto/auth/v1/auth.proto`. Atom implements `AuthService` so a broker can call +it directly, with no adapter service in between. + +| | | +|---|---| +| Source | https://github.com/absmach/fluxmq | +| Path | `proto/auth/v1/auth.proto` | +| Pinned ref | see `REF` in this directory | + +## Why it is byte-identical + +Nothing Atom-specific belongs in this file. Drift from upstream is detected by a +plain `diff`, and a diff can only stay trustworthy if there is nothing expected +to differ — a locally-edited header would mean the check had to know which +differences to forgive, and a check that forgives differences stops catching the +one that matters. Atom's own notes live in this file and in +`AGENTS.md § Broker auth callout`. + +## Checking for drift + +```bash +scripts/check-vendored-proto.sh +``` + +CI runs the same script. It fetches the pinned ref from GitHub and diffs. + +## When it fails + +A failure means upstream changed the contract Atom implements. That is +information, not a chore — read the diff before syncing: + +- **Comments or new optional fields** — re-vendor, bump `REF`, done. +- **A changed `package` line** — the gRPC path a broker dials is derived from + it, so this is a breaking wire change. Atom, the broker, and any adapter + service must move together; see the deployment note in `AGENTS.md`. +- **Renamed or renumbered fields** — check `src/broker_auth/service.rs` before + re-vendoring. `prost` will happily compile a field that now means something + else. + +## Syncing + +```bash +curl -fsSL "https://raw.githubusercontent.com/absmach/fluxmq/$(cat proto/broker/v1/REF)/proto/auth/v1/auth.proto" \ + -o proto/broker/v1/auth.proto +cargo test --lib broker_auth +``` diff --git a/scripts/check-vendored-proto.sh b/scripts/check-vendored-proto.sh new file mode 100755 index 00000000..d7b3a665 --- /dev/null +++ b/scripts/check-vendored-proto.sh @@ -0,0 +1,57 @@ +#!/usr/bin/env bash +# +# Detect drift between Atom's vendored broker-callout contract and the upstream +# it was copied from. +# +# Atom implements a proto it does not own. Nothing rebuilds the vendored copy, so +# without this check an upstream change is discovered at runtime — as an +# UNIMPLEMENTED from a renamed service, or worse, as a field that still decodes +# but no longer means what Atom thinks it means. +# +# See proto/broker/v1/VENDOR.md for what to do when this fails. + +set -euo pipefail + +readonly VENDOR_DIR="proto/broker/v1" +readonly VENDORED="${VENDOR_DIR}/auth.proto" +readonly REF_FILE="${VENDOR_DIR}/REF" +readonly UPSTREAM_REPO="absmach/fluxmq" +readonly UPSTREAM_PATH="proto/auth/v1/auth.proto" + +if [[ ! -f "${VENDORED}" ]]; then + echo "error: ${VENDORED} not found; run from the repository root" >&2 + exit 2 +fi + +ref="$(tr -d '[:space:]' <"${REF_FILE}")" +if [[ -z "${ref}" ]]; then + echo "error: ${REF_FILE} is empty; it must name a branch, tag, or commit" >&2 + exit 2 +fi + +url="https://raw.githubusercontent.com/${UPSTREAM_REPO}/${ref}/${UPSTREAM_PATH}" +upstream="$(mktemp)" +trap 'rm -f "${upstream}"' EXIT + +if ! curl -fsSL "${url}" -o "${upstream}"; then + echo "error: could not fetch ${url}" >&2 + echo " check that ${REF_FILE} names a ref that exists upstream" >&2 + exit 2 +fi + +if diff -u "${upstream}" "${VENDORED}"; then + echo "vendored proto matches ${UPSTREAM_REPO}@${ref}" + exit 0 +fi + +cat >&2 < Date: Mon, 10 Aug 2026 21:27:11 +0200 Subject: [PATCH 5/6] Correct the package a test's doc comment names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The vendored contract moved to `broker.auth.v1`, but this file still named `fluxmq.auth.v1` — a package no peer serves any more. It was the last reference to the old name left in the repository. Drop the attribution to FluxMQ along with it. Calling the service FluxMQ's is what the rename set out to undo: the contract is the broker's, and Atom serves it for whichever broker dials. --- tests/m29_broker_auth_callout.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/m29_broker_auth_callout.rs b/tests/m29_broker_auth_callout.rs index 3280ed13..9c006b4e 100644 --- a/tests/m29_broker_auth_callout.rs +++ b/tests/m29_broker_auth_callout.rs @@ -1,5 +1,5 @@ -//! DB-gated tests for the broker auth callout — Atom serving FluxMQ's -//! `fluxmq.auth.v1.AuthService` directly, with no adapter service in between. +//! DB-gated tests for the broker auth callout — Atom serving +//! `broker.auth.v1.AuthService` directly, with no adapter service in between. //! //! Run with: //! ```bash From ffed8f78dc6316875dd929fe27f97c2fcf7051ac Mon Sep 17 00:00:00 2001 From: dusan Date: Mon, 10 Aug 2026 23:27:04 +0200 Subject: [PATCH 6/6] Fix formatting Signed-off-by: dusan --- proto/broker/v1/VENDOR.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/proto/broker/v1/VENDOR.md b/proto/broker/v1/VENDOR.md index d7b985a5..8456ffda 100644 --- a/proto/broker/v1/VENDOR.md +++ b/proto/broker/v1/VENDOR.md @@ -4,11 +4,11 @@ `proto/auth/v1/auth.proto`. Atom implements `AuthService` so a broker can call it directly, with no adapter service in between. -| | | -|---|---| -| Source | https://github.com/absmach/fluxmq | -| Path | `proto/auth/v1/auth.proto` | -| Pinned ref | see `REF` in this directory | +| | | +| ---------- | --------------------------------- | +| Source | https://github.com/absmach/fluxmq | +| Path | `proto/auth/v1/auth.proto` | +| Pinned ref | see `REF` in this directory | ## Why it is byte-identical