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 6db146d6..4e55d83a 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 + │ `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) db.rs — pool creation (configurable pool) models/ @@ -200,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. @@ -221,6 +233,83 @@ 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` 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: + +| 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` | +| `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 +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/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/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/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..8456ffda --- /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/proto/broker/v1/auth.proto b/proto/broker/v1/auth.proto new file mode 100644 index 00000000..75963f1d --- /dev/null +++ b/proto/broker/v1/auth.proto @@ -0,0 +1,146 @@ +// Copyright (c) Abstract Machines +// SPDX-License-Identifier: Apache-2.0 + +syntax = "proto3"; + +// 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 +// 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/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 < 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"))); + }; + + // 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. + 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..f8f2987c --- /dev/null +++ b/src/broker_auth/topic.rs @@ -0,0 +1,667 @@ +//! 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() + } +} + +/// 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::*; + + 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()); + } + + // ── 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 e4a46ac2..c8a90023 100644 --- a/src/config.rs +++ b/src/config.rs @@ -75,6 +75,103 @@ 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, + /// 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 +/// 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, + 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") + .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 +637,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 +704,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 +1055,40 @@ 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()?, + topic_allow: crate::broker_auth::TopicAllowList::parse_list(&comma_list( + "ATOM_BROKER_TOPIC_ALLOW", + ))?, + }) +} + 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..9c006b4e --- /dev/null +++ b/tests/m29_broker_auth_callout.rs @@ -0,0 +1,460 @@ +//! DB-gated tests for the broker auth callout — Atom serving +//! `broker.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); +} + +/// 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] +#[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); +}