🥮 mooncakes.io

    moonkafka

    Download zip
    Author
    Version
    0.1.0
    License
    MIT
    Last updated
    22 hours ago
    Downloads
    1

    Dependencies

    #moonkafka

    An open-source Apache Kafka client driver written in MoonBit.

    Target: Apache Kafka 4.x only.

    This driver deliberately supports only the latest Kafka generation — KRaft-based clusters (no ZooKeeper), the modern Kafka protocol, and no legacy broker/version compatibility baggage.

    Status: early development.

    The wire protocol implementation is in progress; the API surface below reflects the intended design and may change before the first release.

    #Why MoonBit + Kafka?

    • MoonBit compiles to small, fast WebAssembly (and native) targets, making it a good fit for lightweight producers/consumers in edge, serverless, and embedded environments.
    • Kafka 4.x is a clean protocol baseline: by dropping ZooKeeper-era and pre-4.x compatibility, the driver stays small and easy to reason about.

    #Features

    Working today:

    • Kafka wire protocol codecs (compact types, zig-zag varints, tagged fields)
    • RecordBatch v2 decoding with CRC32C verification, and encoding for producing
    • ApiVersions v3, Produce v11, Metadata v12, ListOffsets v7, Fetch v12
    • Simple producer: per-leader connections, Kafka-compatible murmur2 key-partitioning (round-robin for keyless messages), metadata refresh on leadership changes
    • Simple consumer: per-leader connections, all partitions, in-memory offsets, earliest/latest start, metadata refresh on leadership changes

    Planned:

    • Batched/async producer
    • Consumer groups with the new KIP-848 consumer rebalance protocol
    • Compressed batches (gzip/snappy/lz4/zstd)
    • TLS and SASL authentication

    #Requirements

    • MoonBit toolchain (latest stable)
    • An Apache Kafka 4.x cluster (KRaft mode). Older brokers are not supported and will not be.

    #Installation

    moon add daqing/moonkafka

    #Quick start

    ///|
    async fn main {
      // Produce
      let producer = @moonkafka.Producer::connect(
        host="127.0.0.1",
        port=9092,
        topic="events",
      )
      defer producer.close()
      let offset = producer.send(
        key=@utf8.encode("k1"),
        value=@utf8.encode("hello"),
      )
      println("produced at offset \{offset}")

      // Consume
      let consumer = @moonkafka.Consumer::connect(
        host="127.0.0.1",
        port=9092,
        topic="events",
        start_from=@moonkafka.StartFrom::Earliest,
      )
      defer consumer.close()
      for ;; {
        for record in consumer.poll() {
          println("offset=\{record.offset} value=\{record.value}")
        }
      }
    }

    A runnable example lives in cmd/main — it consumes all partitions of a topic and prints records until interrupted:

    moon run cmd/main -- consume events [host] [port]

    It can also produce a single message:

    moon run cmd/main -- produce events "hello" [key] [host] [port]

    Note: the socket layer is native-backend only (moonbitlang/async), so the module targets native.

    #Development

    moon build          # build the library
moon test           # run tests (blackbox + whitebox)
moon fmt            # format code
moon info           # regenerate package interfaces (.mbti)

    #License

    MIT

    BrokerError

    source
    pub(all) suberror BrokerError {
      BrokerError(Int, String)
    }

    An error code reported by a broker in a response. message carries call-site context (which request, which topic/partition) on top of the protocol-level code.

    BrokerError::is_retriable

    source
    fn BrokerError::is_retriable(self : BrokerError) -> Bool

    Whether the protocol marks this error as retriable.

    BrokerError::name

    source
    fn BrokerError::name(self : BrokerError) -> String

    Protocol name for the error code, e.g. "OFFSET_OUT_OF_RANGE".

    ProtocolError

    source
    pub(all) suberror ProtocolError {
      ProtocolError(String)
    }

    SaslError

    source
    pub(all) suberror SaslError {
      SaslError(String)
    }

    SASL negotiation or authentication failure. The connection is closed.

    TransportError

    source
    pub(all) suberror TransportError {
      ConnectionClosed(String)
      RequestTimeout(String)
    }

    Transport-level failure. The affected connection has been closed when either variant is raised; callers reconnect rather than retry in place.

    Backoff

    source
    pub struct Backoff {
      base_ms : Int
      max_ms : Int
      attempt : Int
    } derive(
    Debug    )

    Stateful backoff schedule: next_ms() grows exponentially per call and applies ±20% wall-clock jitter so retried clients do not synchronize.

    Backoff::new

    source
    fn Backoff::new(base_ms? : Int, max_ms? : Int) -> Backoff raise

    Backoff::next_ms

    source
    fn Backoff::next_ms(self : Backoff) -> Int

    Next delay in milliseconds, advancing the attempt counter.

    Backoff::reset

    source
    fn Backoff::reset(self : Backoff) -> Unit

    BootstrapServers

    source
    pub struct BootstrapServers {
      addrs : Array[HostPort]
      cursor : Int
    } derive(
    Debug    )

    Rotating view over parsed bootstrap server addresses.

    BootstrapServers::new

    source
    fn BootstrapServers::new(addrs : Array[HostPort]) -> BootstrapServers raise

    BootstrapServers::next

    source

    Next address in round-robin order.

    BrokerConnection

    source
    pub struct BrokerConnection {
      tcp : 
    Tcp
      stream : BrokerStream
      client_id : String
      write_lock : 
    Mutex
      read_lock : 
    Mutex
      responses : Map[Int, Bytes]
      correlation_id : Int
      closed : Bool
      in_flight : 
    Semaphore
      throttled_until_ms : Int64
    }

    BrokerConnection::authenticate

    source
    async fn BrokerConnection::authenticate(self : BrokerConnection, sasl : SaslConfig, timeout_ms? : Int) -> Unit

    Run the SASL handshake and authentication exchange on an established connection. OAUTHBEARER and the SCRAM family raise until their implementations land.

    BrokerConnection::check_api_versions

    source
    async fn BrokerConnection::check_api_versions(self : BrokerConnection, timeout_ms? : Int) -> Unit

    BrokerConnection::close

    source
    fn BrokerConnection::close(self : BrokerConnection) -> Unit

    Close the underlying socket. Idempotent; later requests fail with ConnectionClosed, and in-flight ones fail as their reads break.

    BrokerConnection::connect

    source
    async fn BrokerConnection::connect(host : String, port : Int, client_id? : String, max_in_flight? : Int, sasl? : SaslConfig?, timeout_ms? : Int, tls? : TlsClientOptions?) -> BrokerConnection

    BrokerConnection::fetch

    source
    async fn BrokerConnection::fetch(self : BrokerConnection, topic : String, partitions : Array[(PartitionInfo, Int64)], max_wait_ms~ : Int, max_bytes~ : Int, timeout_ms? : Int) -> Array[FetchPartitionResult]

    BrokerConnection::fetch_metadata

    source
    async fn BrokerConnection::fetch_metadata(self : BrokerConnection, topic : String, timeout_ms? : Int) -> Metadata

    BrokerConnection::list_offsets

    source
    async fn BrokerConnection::list_offsets(self : BrokerConnection, topic : String, partitions : Array[PartitionInfo], timestamp : Int64, timeout_ms? : Int) -> Map[Int, Int64]

    BrokerConnection::note_throttle

    source
    fn BrokerConnection::note_throttle(self : BrokerConnection, throttle_ms : Int) -> Unit

    Record the broker's latest throttle hint, keeping the furthest deadline.

    BrokerConnection::produce

    source
    async fn BrokerConnection::produce(self : BrokerConnection, topic : String, partitions : Array[(Int, Bytes)], acks~ : Int, timeout_ms~ : Int) -> Array[ProducePartitionResult]

    BrokerConnection::request

    source
    async fn BrokerConnection::request(self : BrokerConnection, api_key : Int, api_version : Int, body : Bytes, timeout_ms? : Int) -> 
    Decoder

    Send one request and wait for its response. At most max_in_flight requests run concurrently; the rest queue on the semaphore. On timeout or any transport failure the connection is closed.

    BrokerConnection::request_raw

    source
    async fn BrokerConnection::request_raw(self : BrokerConnection, api_key : Int, api_version : Int, body : Bytes, timeout_ms? : Int, flexible? : Bool) -> 
    Decoder

    Like request, but for APIs whose chosen version is not flexible (request header v1, response header v0): SaslHandshake v1 today.

    BrokerInfo

    source
    pub struct BrokerInfo {
      node_id : Int
      host : String
      port : Int
    }

    BrokerStream

    source
    pub(all) enum BrokerStream {
      Plain(
    Tcp    )
      Secure(
    Tls    )
    }

    Byte stream to a broker: plaintext TCP, or TLS layered over TCP.

    BrokerStream::read_exactly

    source
    async fn BrokerStream::read_exactly(self : BrokerStream, n : Int) -> Bytes

    BrokerStream::shutdown

    source
    fn BrokerStream::shutdown(self : BrokerStream) -> Unit

    Release the stream. The TLS layer must be freed before the raw socket underneath it, per the @tls requirements.

    BrokerStream::write

    source
    async fn BrokerStream::write(self : BrokerStream, data : Bytes) -> Unit

    CommonConfig

    source
    pub struct CommonConfig {
      bootstrap_servers : Array[String]
      client_id : String
      request_timeout_ms : Int
      connection_max_idle_ms : Int
      retries : Int
      retry_backoff_ms : Int
      retry_backoff_max_ms : Int
      security_protocol : SecurityProtocol
      sasl : SaslConfig?
      tls : TlsClientOptions?
    } derive(
    Debug    )

    Settings shared by every client surface.

    CommonConfig::bootstrap_addresses

    source
    fn CommonConfig::bootstrap_addresses(self : CommonConfig) -> Array[HostPort] raise ProtocolError

    Parse every bootstrap entry; raises on the first malformed one.

    CommonConfig::new

    source
    fn CommonConfig::new(bootstrap_servers : Array[String], request_timeout_ms? : Int, security_protocol? : SecurityProtocol, sasl? : SaslConfig?, tls? : TlsClientOptions?) -> CommonConfig raise

    Defaults mirror the Java client where sensible: 30s request timeout, 9-minute idle connection limit. Retries are bounded at 10 for now — the Java default is effectively unbounded but is gated by delivery.timeout.ms, which the batching producer (Phase 3) introduces.

    CommonConfig::validate

    source
    fn CommonConfig::validate(self : CommonConfig) -> Unit raise ProtocolError

    Reject values that would make a client misbehave; called by new.

    Consumer

    source
    pub struct Consumer {
      topic : String
      start_from : StartFrom
      request_timeout_ms : Int
      bootstrap : BootstrapServers
      sasl : SaslConfig?
      meta_conn : BrokerConnection
      brokers : Map[Int, BrokerInfo]
      leader_conns : Map[Int, BrokerConnection]
      partitions : Array[PartitionState]
      closed : Bool
    }

    Consumer::close

    source
    fn Consumer::close(self : Consumer) -> Unit

    Consumer::connect

    source
    async fn Consumer::connect(host~ : String, port~ : Int, topic~ : String, start_from? : StartFrom) -> Consumer

    Connect to a bootstrap broker, negotiate API versions, resolve the topic's partitions and their leaders, and initialize fetch offsets.

    Consumer::connect_with_config

    source
    async fn Consumer::connect_with_config(config : ConsumerConfig) -> Consumer

    Connect using explicit configuration, which is validated first. The meta connection lands on the first bootstrap server that accepts.

    Consumer::poll

    source
    async fn Consumer::poll(self : Consumer, max_wait_ms? : Int, max_bytes? : Int) -> Array[Record]

    Poll every partition once. Returns the decoded records (possibly empty; the broker long-polls up to max_wait_ms per fetch). On offset or leadership errors the consumer recovers by refreshing metadata or re-resolving offsets on the next poll.

    ConsumerConfig

    source
    pub struct ConsumerConfig {
      common : CommonConfig
      topic : String
      start_from : StartFrom
    } derive(
    Debug    )

    ConsumerConfig::new

    source
    fn ConsumerConfig::new(bootstrap_servers : Array[String], topic : String, start_from? : StartFrom, request_timeout_ms? : Int, security_protocol? : SecurityProtocol, sasl? : SaslConfig?, tls? : TlsClientOptions?) -> ConsumerConfig raise

    Deadline

    source
    pub struct Deadline {
      at_ms : Int64
    } derive(
    Debug    )

    A point in time measured against the @async wall clock.

    Deadline::after_ms

    source
    fn Deadline::after_ms(ms : Int64) -> Deadline

    Deadline::expired

    source
    fn Deadline::expired(self : Deadline) -> Bool

    Deadline::remaining_ms

    source
    fn Deadline::remaining_ms(self : Deadline) -> Int64

    Milliseconds until expiry; negative once past.

    FetchPartitionResult

    source
    pub struct FetchPartitionResult {
      partition : Int
      error_code : Int
      high_watermark : Int64
      records : Array[Record]
    }

    HostPort

    source
    pub(all) struct HostPort {
      host : String
      port : Int
    } derive(Compare, Eq, Hash, 
    Debug    )

    One "host:port" (or "[ipv6]:port") entry of the bootstrap list.

    HostPort::parse

    source
    fn HostPort::parse(entry : String) -> HostPort raise ProtocolError

    Parse a single bootstrap entry. The port defaults to 9092 when omitted; bare IPv6 addresses must be bracketed.

    Metadata

    source
    pub struct Metadata {
      brokers : Map[Int, BrokerInfo]
      topics : Array[TopicMetadata]
    }

    PartitionInfo

    source
    pub struct PartitionInfo {
      index : Int
      leader : Int
      leader_epoch : Int
    }

    PartitionState

    source
    type PartitionState

    ProducePartitionResult

    source
    pub struct ProducePartitionResult {
      partition : Int
      error_code : Int
      base_offset : Int64
    }

    Producer

    source
    pub struct Producer {
      topic : String
      acks : Int
      timeout_ms : Int
      retries : Int
      retry_backoff_ms : Int
      retry_backoff_max_ms : Int
      bootstrap : BootstrapServers
      sasl : SaslConfig?
      use_tls : Bool
      tls_options : TlsClientOptions?
      meta_conn : BrokerConnection
      brokers : Map[Int, BrokerInfo]
      leader_conns : Map[Int, BrokerConnection]
      partitions : Array[PartitionInfo]
      round_robin : Int
      closed : Bool
    }

    Producer::close

    source
    fn Producer::close(self : Producer) -> Unit

    Producer::connect

    source
    async fn Producer::connect(host~ : String, port~ : Int, topic~ : String, acks? : Int, timeout_ms? : Int) -> Producer

    Connect to a bootstrap broker, negotiate API versions, resolve the topic's partitions and their leaders, and open a connection per leader. acks is 1 (leader acknowledgement) or -1 (all in-sync replicas); acks=0 is not supported because this client always reads the response.

    Producer::connect_with_config

    source
    async fn Producer::connect_with_config(config : ProducerConfig) -> Producer

    Connect using explicit configuration, which is validated first. The meta connection lands on the first bootstrap server that accepts.

    Producer::send

    source
    async fn Producer::send(self : Producer, key? : Bytes, value? : Bytes, timestamp? : Int64) -> Int64

    Send one message and return the offset assigned by the broker. timestamp defaults to the current time (ms since epoch). Leadership changes and transport failures trigger recovery and a backoff-paced retry, up to retries attempts after the first.

    ProducerConfig

    source
    pub struct ProducerConfig {
      common : CommonConfig
      topic : String
      acks : Int
    } derive(
    Debug    )

    ProducerConfig::new

    source
    fn ProducerConfig::new(bootstrap_servers : Array[String], topic : String, acks? : Int, request_timeout_ms? : Int, security_protocol? : SecurityProtocol, sasl? : SaslConfig?, tls? : TlsClientOptions?) -> ProducerConfig raise

    acks is 1 (leader acknowledgement) or -1 (all in-sync replicas); 0 becomes valid with the Phase 3 sender that does not read responses.

    Record

    source
    pub(all) struct Record {
      offset : Int64
      timestamp : Int64
      key : Bytes?
      value : Bytes?
    }

    SaslAuthenticateResult

    source
    pub(all) struct SaslAuthenticateResult {
      error_code : Int
      error_message : String?
      auth_bytes : Bytes?
      session_lifetime_ms : Int64
    }

    SaslConfig

    source
    pub(all) struct SaslConfig {
      mechanism : SaslMechanism
      username : String
      password : String
    } derive(
    Debug    )

    SaslHandshakeResult

    source
    pub(all) struct SaslHandshakeResult {
      error_code : Int
      mechanisms : Array[String]
    }

    SaslMechanism

    source
    pub(all) enum SaslMechanism {
      Plain
      ScramSha256
      ScramSha512
      OAuthBearer
    } derive(
    Debug    )

    SASL mechanism. OAUTHBEARER token callbacks arrive with the Phase 1 SASL work; PLAIN and SCRAM use username/password.

    ScramServerFirst

    source
    pub(all) struct ScramServerFirst {
      nonce : String
      salt : Bytes
      iterations : Int
    } derive(
    Debug    )

    ScramServerFirst::parse

    source
    fn ScramServerFirst::parse(msg : String) -> ScramServerFirst raise

    Parse a server-first-message into its nonce (which must extend the client nonce), salt, and iteration count.

    SecurityProtocol

    source
    pub(all) enum SecurityProtocol {
      Plaintext
      Ssl
      SaslPlaintext
      SaslSsl
    } derive(
    Debug    )

    Transport security. SASL layers authenticate right after connecting; only plaintext is wired up so far.

    StartFrom

    source
    pub(all) enum StartFrom {
      Earliest
      Latest
    } derive(
    Debug    )

    TlsClientOptions

    source
    pub(all) struct TlsClientOptions {
      server_name : String
      verify_certificates : Bool
      ca_pem_file : String?
    } derive(
    Debug    )

    TLS settings for a broker connection. server_name drives SNI and certificate verification; ca_pem_file overrides the system trust roots with a custom CA bundle.

    TlsClientOptions::new

    source
    fn TlsClientOptions::new(server_name : String, verify_certificates? : Bool, ca_pem_file? : String?) -> TlsClientOptions

    TopicMetadata

    source
    pub struct TopicMetadata {
      name : String
      partitions : Array[PartitionInfo]
    }

    Uuid

    source
    pub struct Uuid {
      bytes : Bytes
    } derive(Compare, Eq, Hash, 
    Debug    )

    impl Show for Uuid

    Uuid::from_bytes

    source
    fn Uuid::from_bytes(bytes : Bytes) -> Uuid raise 
    DecodeError

    Wrap exactly 16 bytes; anything else is a wire-level violation.

    Uuid::parse

    source

    Parse the canonical 22-char form (as produced by Show). The zero padding in the trailing character's low bits is ignored, like Java's decoder.

    Uuid::to_bytes

    source
    fn Uuid::to_bytes(self : Uuid) -> Bytes

    Uuid::zero

    source
    fn Uuid::zero() -> Uuid

    The all-zero uuid, Kafka's sentinel for "no id".

    API_API_VERSIONS

    source
    let API_API_VERSIONS : Int

    API_FETCH

    source
    let API_FETCH : Int

    API_LIST_OFFSETS

    source
    let API_LIST_OFFSETS : Int

    API_METADATA

    source
    let API_METADATA : Int

    API_PRODUCE

    source
    let API_PRODUCE : Int

    backoff_ms

    source
    fn backoff_ms(base_ms : Int, max_ms : Int, attempt : Int) -> Int

    Pure exponential growth: base_ms shifted by attempt, capped at max_ms (and at a shift ceiling so large attempts cannot overflow).

    connect_bootstrap

    source
    async fn connect_bootstrap(servers : BootstrapServers, client_id : String, timeout_ms? : Int, sasl? : SaslConfig?, use_tls? : Bool, tls? : TlsClientOptions?) -> BrokerConnection

    Dial the bootstrap servers in rotating order; the first connection that succeeds wins. Raises the last failure if every entry refuses.

    decode_api_versions_response

    source
    fn decode_api_versions_response(d : 
    Decoder    ) -> Int raise

    Check the ApiVersions v3 response; response header is v0 (no tag buffer). Raises unless the broker supports the API versions this client uses. Returns the throttle hint in milliseconds.

    decode_fetch_response

    source
    fn decode_fetch_response(d : 
    Decoder    ) -> (Array[FetchPartitionResult], Int) raise

    Decode a Fetch v12 response (after the v1 response header).

    decode_list_offsets_response

    source
    fn decode_list_offsets_response(d : 
    Decoder    ) -> (Map[Int, Int64], Int) raise

    Decode a ListOffsets v7 response; returns partition index -> offset plus the throttle hint in milliseconds.

    decode_metadata_response

    source
    fn decode_metadata_response(d : 
    Decoder    ) -> (Metadata, Int) raise

    Decode a Metadata v12 response (after the v1 response header). Returns the metadata and the throttle hint in milliseconds.

    decode_produce_response

    source
    fn decode_produce_response(d : 
    Decoder    ) -> (Array[ProducePartitionResult], Int) raise

    Decode a Produce v11 response (after the v1 response header).

    decode_record_batches

    source
    fn decode_record_batches(data : Bytes) -> Array[Record] raise 
    DecodeError

    Decode all record batches concatenated in data (as found in a Fetch response's records field). Compressed and control batches are skipped. A truncated trailing batch (possible when max_bytes splits a batch) is silently dropped; the caller refetches from the last good offset + 1.

    decode_sasl_authenticate_response

    source
    fn decode_sasl_authenticate_response(d : 
    Decoder    ) -> SaslAuthenticateResult raise

    decode_sasl_handshake_response

    source
    fn decode_sasl_handshake_response(d : 
    Decoder    ) -> SaslHandshakeResult raise

    SaslHandshake v1 response: non-flexible, so the mechanisms array uses an int32 count and int16-length strings.

    encode_api_versions_request

    source
    fn encode_api_versions_request() -> Bytes

    encode_fetch_request

    source
    fn encode_fetch_request(topic : String, partitions : Array[(PartitionInfo, Int64)], max_wait_ms : Int, max_bytes : Int) -> Bytes

    encode_list_offsets_request

    source
    fn encode_list_offsets_request(topic : String, partitions : Array[PartitionInfo], timestamp : Int64) -> Bytes

    encode_metadata_request

    source
    fn encode_metadata_request(topic : String) -> Bytes

    encode_produce_request

    source
    fn encode_produce_request(topic : String, partitions : Array[(Int, Bytes)], acks : Int, timeout_ms : Int) -> Bytes

    Encode a Produce v11 request: one topic, one record batch per partition.

    encode_record_batch

    source
    fn encode_record_batch(records : Array[Record]) -> Bytes

    Encode records into a single RecordBatch v2 (magic = 2), uncompressed, base offset 0, create-time timestamps. Record timestamps are absolute ms since epoch; offset deltas are assigned sequentially from 0.

    encode_request

    source
    fn encode_request(api_key : Int, api_version : Int, correlation_id : Int, client_id : String, body : Bytes, flexible? : Bool) -> Bytes

    Encode a full request frame: INT32 size prefix + header v2 + body. Header v2 keeps client_id as a legacy NULLABLE_STRING (see RequestHeader.json) followed by an empty tag buffer.

    encode_sasl_authenticate_request

    source
    fn encode_sasl_authenticate_request(auth_bytes : Bytes) -> Bytes

    SaslAuthenticate v2 body: auth bytes as a compact byte array.

    encode_sasl_handshake_request

    source
    fn encode_sasl_handshake_request(mechanism : String) -> Bytes

    SaslHandshake v1 body: the mechanism name as a NON-compact string.

    error_name

    source
    fn error_name(code : Int) -> String

    Protocol name for a broker error code; unknown codes render as "UNKNOWN(code)".

    error_retriable

    source
    fn error_retriable(code : Int) -> Bool

    Whether the protocol marks the error code as retriable; unknown codes are treated as not retriable.

    scram_auth_message

    source
    fn scram_auth_message(client_first_bare : String, server_first : String, client_final_without_proof : String) -> String

    AuthMessage = client-first-bare "," server-first "," client-final-without-proof.

    scram_client_proof_b64

    source
    fn scram_client_proof_b64(sha512 : Bool, salted_password : Bytes, auth_message : String) -> String

    ClientProof = ClientKey XOR ClientSignature, base64-encoded.

    scram_escape_username

    source
    fn scram_escape_username(username : String) -> String

    Escape the username per the SCRAM spec: '=' and ',' are sent as =3D and =2C. (Full SASLprep normalization is not applied; usernames made of printable ASCII are unaffected.)

    scram_pbkdf2_hmac

    source
    fn[H : 
    CryptoHasher    ] scram_pbkdf2_hmac(h : H, password : Bytes, salt : Bytes, iterations : Int, dk_len : Int) -> Bytes

    PBKDF2-HMAC (RFC 2898) producing exactly dk_len bytes; SCRAM always uses dk_len equal to the hash output length, i.e. a single block.

    scram_salted_password

    source
    fn scram_salted_password(sha512 : Bool, password : Bytes, salt : Bytes, iterations : Int) -> Bytes

    SaltedPassword = Hi(password, salt, iterations), i.e. PBKDF2-HMAC.

    scram_server_final

    source
    fn scram_server_final(sha512 : Bool, salted_password : Bytes, auth_message : String) -> String

    The broker's server signature, verified against the server-final message; also used by the fake-broker tests from the server side.

    scram_server_signature_b64

    source
    fn scram_server_signature_b64(sha512 : Bool, salted_password : Bytes, auth_message : String) -> String

    ServerSignature = HMAC(ServerKey, AuthMessage), base64-encoded; the client verifies the broker's server-final "v=" against it.

    scram_verify_client_proof

    source
    fn scram_verify_client_proof(sha512 : Bool, salted_password : Bytes, auth_message : String, proof_b64 : String) -> Bool

    Server-side check of a client proof: recompute the stored key from the known salted password and confirm proof XOR ClientSignature hashes to it.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io