krabka_raft/server/
authorization.rs

1//! Per-request authorization on the controller listener.
2//!
3//! Kafka's `ControllerApis` authorizes every request with the cluster
4//! operation that its api needs, and a denial becomes the
5//! `getErrorResponse` of that api with `CLUSTER_AUTHORIZATION_FAILED`. This
6//! module holds that table for the apis that the controller listener answers
7//! itself, and the refusal body of each one.
8//!
9//! The apis that the KIP-919 Admin router sends to a broker handler are not
10//! in the table: each of those handlers checks its own operation. The
11//! exception is `AssignReplicasToDirs`, whose broker handler takes no
12//! principal, so the table holds it. `ApiVersions` needs no grant.
13
14use {
15    crate::{
16        ClusterOperation,
17        error::RaftError,
18        kraft::transport::{PeerApi, api_key},
19        wire::{
20            API_KEY_DELEGATION_TOKEN_MUTATION, API_KEY_METADATA_FETCH, API_KEY_SUBMIT_CHANGE,
21            KrabkaMetadataFetchResponse, KrabkaSubmitChangeResponse,
22            PRIVATE_CLUSTER_AUTHORIZATION_FAILED,
23        },
24    },
25    bytes::{Bytes, BytesMut},
26    krabka_protocol::{
27        Decode, Encode,
28        owned::{
29            self,
30            fetch_request::FetchRequest,
31            fetch_response::{FetchResponse, FetchableTopicResponse, PartitionData},
32        },
33        records::RecordsPayload,
34    },
35};
36
37/// Kafka's `CLUSTER_AUTHORIZATION_FAILED`.
38const CLUSTER_AUTHORIZATION_FAILED: i16 = 31;
39
40/// Kafka's message for `CLUSTER_AUTHORIZATION_FAILED`, which the apis with an
41/// error message field carry.
42const CLUSTER_AUTHORIZATION_FAILED_MESSAGE: &str = "Cluster authorization failed.";
43
44/// Kafka's `FetchResponse.INVALID_HIGH_WATERMARK`.
45const INVALID_HIGH_WATERMARK: i64 = -1;
46
47/// The first `Fetch` version whose error response has no partition rows.
48const FETCH_TOP_LEVEL_ERROR_ONLY_VERSION: i16 = 13;
49
50/// The cluster operation that `api_key` needs on the controller listener, or
51/// `None` when the listener does not authorize it itself.
52///
53/// The operations are Kafka's, from `ControllerApis`: `handleFetch`,
54/// `handleFetchSnapshot`, `handleVote`, `handleBeginQuorumEpoch`,
55/// `handleEndQuorumEpoch`, `handleControllerRegistration`,
56/// `handleUpdateRaftVoter` and `handleAssignReplicasToDirs` need
57/// `CLUSTER_ACTION`. `handleAddRaftVoter`, `handleRemoveRaftVoter` and
58/// `handleDescribeCluster` need `ALTER`. `handleDescribeQuorum` needs
59/// `DESCRIBE`. The krabka-private metadata apis have no Kafka counterpart.
60/// They read or write the metadata log for another node, so they need
61/// `CLUSTER_ACTION`.
62pub(super) const fn required_operation(api_key: i16) -> Option<ClusterOperation> {
63    match api_key {
64        peer if PeerApi::from_api_key(peer).is_some() => Some(ClusterOperation::ClusterAction),
65        owned::controller_registration_request::API_KEY
66        | owned::update_raft_voter_request::API_KEY
67        | owned::assign_replicas_to_dirs_request::API_KEY
68        | API_KEY_SUBMIT_CHANGE
69        | API_KEY_METADATA_FETCH
70        | API_KEY_DELEGATION_TOKEN_MUTATION => Some(ClusterOperation::ClusterAction),
71        owned::add_raft_voter_request::API_KEY
72        | owned::remove_raft_voter_request::API_KEY
73        | owned::describe_cluster_request::API_KEY => Some(ClusterOperation::Alter),
74        owned::describe_quorum_request::API_KEY => Some(ClusterOperation::Describe),
75        _ => None,
76    }
77}
78
79/// The refusal body of a request that [`required_operation`] names and the
80/// connection principal does not hold.
81///
82/// Each Kafka api answers in the shape of its own `getErrorResponse`. The
83/// krabka-private apis answer their own response with
84/// [`PRIVATE_CLUSTER_AUTHORIZATION_FAILED`].
85///
86/// # Errors
87///
88/// Returns an error when a `Fetch` body does not decode, or when the
89/// response does not encode at `version`.
90pub(super) fn refusal(api_key: i16, version: i16, body: &[u8]) -> Result<Bytes, RaftError> {
91    let mut out = BytesMut::new();
92    let message = || Some(CLUSTER_AUTHORIZATION_FAILED_MESSAGE.to_owned());
93    match api_key {
94        api_key::FETCH => fetch_refusal(version, body)?.encode(&mut out, version)?,
95        api_key::VOTE => krabka_macros::authorization_refusal!(
96            owned::vote_response::VoteResponse,
97            out,
98            version,
99            CLUSTER_AUTHORIZATION_FAILED
100        ),
101        api_key::BEGIN_QUORUM_EPOCH => {
102            krabka_macros::authorization_refusal!(
103                owned::begin_quorum_epoch_response::BeginQuorumEpochResponse,
104                out,
105                version,
106                CLUSTER_AUTHORIZATION_FAILED
107            );
108        }
109        api_key::END_QUORUM_EPOCH => krabka_macros::authorization_refusal!(
110            owned::end_quorum_epoch_response::EndQuorumEpochResponse,
111            out,
112            version,
113            CLUSTER_AUTHORIZATION_FAILED
114        ),
115        api_key::FETCH_SNAPSHOT => krabka_macros::authorization_refusal!(
116            owned::fetch_snapshot_response::FetchSnapshotResponse,
117            out,
118            version,
119            CLUSTER_AUTHORIZATION_FAILED
120        ),
121        owned::assign_replicas_to_dirs_request::API_KEY => {
122            krabka_macros::authorization_refusal!(
123                owned::assign_replicas_to_dirs_response::AssignReplicasToDirsResponse,
124                out,
125                version,
126                CLUSTER_AUTHORIZATION_FAILED
127            );
128        }
129        owned::controller_registration_request::API_KEY => krabka_macros::authorization_refusal!(
130            owned::controller_registration_response::ControllerRegistrationResponse,
131            out,
132            version,
133            CLUSTER_AUTHORIZATION_FAILED,
134            { message() }
135        ),
136        owned::update_raft_voter_request::API_KEY => {
137            krabka_macros::authorization_refusal!(
138                owned::update_raft_voter_response::UpdateRaftVoterResponse,
139                out,
140                version,
141                CLUSTER_AUTHORIZATION_FAILED
142            );
143        }
144        owned::add_raft_voter_request::API_KEY => {
145            krabka_macros::authorization_refusal!(
146                owned::add_raft_voter_response::AddRaftVoterResponse,
147                out,
148                version,
149                CLUSTER_AUTHORIZATION_FAILED,
150                { message() }
151            );
152        }
153        owned::remove_raft_voter_request::API_KEY => krabka_macros::authorization_refusal!(
154            owned::remove_raft_voter_response::RemoveRaftVoterResponse,
155            out,
156            version,
157            CLUSTER_AUTHORIZATION_FAILED,
158            { message() }
159        ),
160        owned::describe_cluster_request::API_KEY => krabka_macros::authorization_refusal!(
161            owned::describe_cluster_response::DescribeClusterResponse,
162            out,
163            version,
164            CLUSTER_AUTHORIZATION_FAILED,
165            { message() }
166        ),
167        owned::describe_quorum_request::API_KEY => krabka_macros::authorization_refusal!(
168            owned::describe_quorum_response::DescribeQuorumResponse,
169            out,
170            version,
171            CLUSTER_AUTHORIZATION_FAILED,
172            { message() }
173        ),
174        API_KEY_SUBMIT_CHANGE | API_KEY_DELEGATION_TOKEN_MUTATION => {
175            let mut private = Vec::new();
176            KrabkaSubmitChangeResponse {
177                error_code: PRIVATE_CLUSTER_AUTHORIZATION_FAILED,
178                leader_hint: -1,
179                result: Bytes::new(),
180            }
181            .encode_v0(&mut private)?;
182            return Ok(Bytes::from(private));
183        }
184        API_KEY_METADATA_FETCH => {
185            let mut private = Vec::new();
186            KrabkaMetadataFetchResponse {
187                error_code: PRIVATE_CLUSTER_AUTHORIZATION_FAILED,
188                leader_hint: -1,
189                leader_epoch: -1,
190                log_start_offset: -1,
191                high_watermark: -1,
192                quorum_high_watermark: -1,
193                snapshot_id: None,
194                records: Bytes::new(),
195            }
196            .encode_v0(&mut private)?;
197            return Ok(Bytes::from(private));
198        }
199        _ => {
200            return Err(RaftError::Protocol(
201                krabka_protocol::ProtocolError::InvalidValue(
202                    "no controller-listener refusal for this api key",
203                ),
204            ));
205        }
206    }
207    Ok(out.freeze())
208}
209
210/// Kafka's `FetchRequest.getErrorResponse`, refused with
211/// `CLUSTER_AUTHORIZATION_FAILED`.
212fn fetch_refusal(version: i16, body: &[u8]) -> Result<FetchResponse, RaftError> {
213    let mut cursor = body;
214    let request = FetchRequest::decode(&mut cursor, version)?;
215    Ok(fetch_error_response(
216        &request,
217        version,
218        CLUSTER_AUTHORIZATION_FAILED,
219    ))
220}
221
222/// Kafka's `FetchRequest.getErrorResponse`: `error_code` and the session id of
223/// `request` at the top level.
224///
225/// Before v13 every requested partition also carries the code, as
226/// `FetchResponse.partitionResponse` builds it: an invalid high watermark and
227/// empty, non-null records.
228#[must_use]
229pub fn fetch_error_response(
230    request: &FetchRequest,
231    version: i16,
232    error_code: i16,
233) -> FetchResponse {
234    let responses = if version < FETCH_TOP_LEVEL_ERROR_ONLY_VERSION {
235        request
236            .topics
237            .iter()
238            .map(|topic| FetchableTopicResponse {
239                topic: topic.topic.clone(),
240                topic_id: topic.topic_id,
241                partitions: topic
242                    .partitions
243                    .iter()
244                    .map(|partition| PartitionData {
245                        partition_index: partition.partition,
246                        error_code,
247                        high_watermark: INVALID_HIGH_WATERMARK,
248                        records: Some(RecordsPayload::default()),
249                        ..Default::default()
250                    })
251                    .collect(),
252                ..Default::default()
253            })
254            .collect()
255    } else {
256        Vec::new()
257    };
258    FetchResponse {
259        error_code,
260        session_id: request.session_id,
261        responses,
262        ..Default::default()
263    }
264}
265
266#[cfg(test)]
267mod tests;
krabka_raft/server/authorization.rs

krabka_raft/server/
authorization.rs
