krabka_raft/
handshake.rs

1//! Pluggable inbound handshake for the controller listener.
2//!
3//! This hook lets the broker terminate TLS and SASL on every accepted
4//! controller-listener connection before the raft frames start to flow. The
5//! trait abstraction keeps `krabka-raft` free of any dependency on
6//! `krabka-broker` and `krabka-security`.
7
8use {bytes::Bytes, krabka_client_core::ClientDuplex, std::sync::Arc, thiserror::Error};
9
10mod stream;
11
12pub use self::stream::{HandshakeClock, HandshakeStream};
13
14/// Authenticated controller-listener connection plus request-level grants.
15pub struct RaftConnection {
16    /// The raft connection handler is generic over `AsyncRead + AsyncWrite +
17    /// Unpin + Send + 'static`, so a `Box<dyn ClientDuplex>` plugs in directly.
18    pub stream: Box<dyn ClientDuplex>,
19    /// Authenticated Kafka principal. `None` represents a PLAINTEXT or
20    /// one-way TLS connection with the normal `ANONYMOUS` identity.
21    pub principal: Option<krabka_security::Principal>,
22    /// Whether SCRAM authenticated with a delegation token rather than a
23    /// regular credential.
24    pub authenticated_via_token: bool,
25    /// Absolute credential deadline, enforced before controller request dispatch.
26    /// `None` means this credential has no finite session deadline.
27    pub expires_at_ms: Option<i64>,
28    /// The cluster grants of the connection principal. The listener asks it
29    /// once for each request, as Kafka's `ControllerApis` does.
30    pub grants: Arc<dyn ClusterGrants>,
31}
32
33/// An operation on the `Cluster("kafka-cluster")` resource that a
34/// controller-listener api needs.
35#[derive(Clone, Copy, Debug, PartialEq, Eq)]
36pub enum ClusterOperation {
37    /// `CLUSTER_ACTION`: the raft, registration and krabka-private metadata
38    /// apis.
39    ClusterAction,
40    /// `ALTER`: `AddRaftVoter`, `RemoveRaftVoter` and `DescribeCluster`.
41    Alter,
42    /// `DESCRIBE`: `DescribeQuorum`.
43    Describe,
44}
45
46/// Decides whether the principal of one controller-listener connection holds
47/// an operation on the cluster resource.
48///
49/// The listener calls [`ClusterGrants::allows`] for every request, so an ACL
50/// change applies to the next request of an open connection.
51pub trait ClusterGrants: Send + Sync {
52    fn allows(&self, operation: ClusterOperation) -> bool;
53
54    /// The KIP-430 bitfield of the operations the principal holds on the
55    /// cluster resource: bit `code` is set for each allowed operation of
56    /// Kafka's `AclEntry.supportedOperations(CLUSTER)`, where `code` is the
57    /// operation's `AclOperation` wire code. `DescribeCluster` answers it
58    /// when the request sets `include_cluster_authorized_operations`.
59    fn cluster_authorized_operations(&self) -> i32;
60}
61
62/// The KIP-430 bitfield of every operation Kafka's
63/// `AclEntry.supportedOperations(CLUSTER)` names: `CREATE` (5), `ALTER` (7),
64/// `DESCRIBE` (8), `CLUSTER_ACTION` (9), `DESCRIBE_CONFIGS` (10),
65/// `ALTER_CONFIGS` (11) and `IDEMPOTENT_WRITE` (12).
66const ALL_CLUSTER_OPERATIONS: i32 =
67    (1 << 5) | (1 << 7) | (1 << 8) | (1 << 9) | (1 << 10) | (1 << 11) | (1 << 12);
68
69/// The grants of a listener that installs no handshake: every operation is
70/// allowed. The broker always installs a handshake, so only a controller
71/// without a broker, such as a raft-only test, uses it.
72#[derive(Debug)]
73pub struct AllowAllGrants;
74
75impl ClusterGrants for AllowAllGrants {
76    fn allows(&self, _operation: ClusterOperation) -> bool { true }
77
78    fn cluster_authorized_operations(&self) -> i32 { ALL_CLUSTER_OPERATIONS }
79}
80
81#[krabka_macros::transport_errors]
82#[derive(Debug, Error)]
83pub enum RaftHandshakeError {
84    #[error("protocol: {0}")]
85    Protocol(String),
86}
87
88/// Per-connection handshake hook.
89///
90/// Implementors consume the accepted [`HandshakeStream`] and return one of
91/// two things. On success they return an authenticated `Box<dyn ClientDuplex>`
92/// that carries the raft frames. On failure they return a
93/// `RaftHandshakeError`, and the listener then drops the connection at debug
94/// level. The listener closes a handshake that reads and writes nothing for
95/// `connections.max.idle.ms`, so a slow peer that keeps sending is not closed.
96#[async_trait::async_trait]
97pub trait RaftListenerHandshake: Send + Sync {
98    /// Upgrades one accepted connection.
99    ///
100    /// A SASL listener answers the `ApiVersions` requests that arrive before
101    /// authentication with `api_versions`, which gives the same answer as the
102    /// listener gives after authentication. Kafka's `SocketServer` hands the
103    /// same `apiVersionSupplier` to `SaslServerAuthenticator`.
104    async fn upgrade(
105        &self,
106        stream: HandshakeStream,
107        api_versions: &dyn ControllerApiVersions,
108    ) -> Result<RaftConnection, RaftHandshakeError>;
109}
110
111/// The controller listener's `ApiVersions` answer, for a handshake that
112/// answers `ApiVersions` before the listener gets the connection.
113pub trait ControllerApiVersions: Send + Sync {
114    /// Answers one `ApiVersions` request with the response body. The body goes
115    /// out behind a v0 response header, at the version the body was encoded
116    /// at: the request version, or v0 for a version the listener does not
117    /// serve.
118    ///
119    /// # Errors
120    /// Returns [`RaftHandshakeError::Protocol`] when the body of a served
121    /// version does not decode. Kafka closes such a connection.
122    fn respond(
123        &self,
124        request_version: i16,
125        request_body: &[u8],
126    ) -> Result<Bytes, RaftHandshakeError>;
127}
krabka_raft/handshake.rs

krabka_raft/
handshake.rs
