Finish auth implementation

Author: sbergen

README.md

@@ -58,7 +58,6 @@ Ok(ReceivedMessage("spoke-test", "Hello from spoke!", False))
 ## Development status
 
 ### Missing MQTT 3.1.1 features
-- Auth
 - Persistent sessions across restarts
 
 ### Transport channels

src/spoke.gleam

@@ -23,7 +23,7 @@ pub type TransportOptions {
 }
 
 pub type AuthDetails {
-  AuthDetails(user_name: String, password: Option(BitArray))
+  AuthDetails(username: String, password: Option(BitArray))
 }
 
 pub type ConnectOptions {
@@ -171,16 +171,21 @@ pub fn server_timeout_ms(
 
 pub fn using_auth(
   options: ConnectOptions,
-  details: AuthDetails,
+  username: String,
+  password: Option(BitArray),
 ) -> ConnectOptions {
-  ConnectOptions(..options, authentication: Some(details))
+  ConnectOptions(
+    ..options,
+    authentication: Some(AuthDetails(username, password)),
+  )
 }
 
 /// Starts a new MQTT client with the given options.
 /// Does not connect to the server, until `connect` is called.
 pub fn start(connect_options: ConnectOptions) -> Client {
   start_with_ms_keep_alive(
     connect_options.client_id,
+    connect_options.authentication,
     connect_options.keep_alive_seconds * 1000,
     connect_options.server_timeout_ms,
     connect_options.transport_options,
@@ -288,13 +293,15 @@ fn call_or_disconnect(
 @internal
 pub fn start_with_ms_keep_alive(
   client_id: String,
+  auth: Option(AuthDetails),
   keep_alive_ms: Int,
   server_timeout_ms: Int,
   transport_opts: TransportOptions,
 ) -> Client {
   let updates = process.new_subject()
   let connect = fn() { create_channel(transport_opts) }
-  let config = Config(client_id, keep_alive_ms, server_timeout_ms, connect)
+  let config =
+    Config(client_id, auth, keep_alive_ms, server_timeout_ms, connect)
   let assert Ok(client) =
     actor.start_spec(actor.Spec(fn() { init(config, updates) }, 100, run_client))
   Client(client, updates, config)
@@ -310,6 +317,7 @@ fn create_channel(options: TransportOptions) -> ChannelResult(ByteChannel) {
 type Config {
   Config(
     client_id: String,
+    auth: Option(AuthDetails),
     keep_alive: Int,
     server_timeout: Int,
     connect: fn() -> ChannelResult(ByteChannel),
@@ -452,6 +460,11 @@ fn handle_connect(
     return: actor.continue(state),
   )
 
+  let auth =
+    option.map(state.config.auth, fn(auth) {
+      packet.AuthOptions(auth.username, auth.password)
+    })
+
   let will =
     option.map(will, fn(will) {
       let data = packet.MessageData(will.topic, will.payload, will.retain)
@@ -466,6 +479,7 @@ fn handle_connect(
       clean_session,
       config.keep_alive,
       config.server_timeout,
+      auth,
       will,
     )
 

src/spoke/internal/connection.gleam

@@ -40,7 +40,6 @@ pub type Update {
   ReceivedPacket(incoming.Packet)
 }
 
-// TODO: Add auth, will, etc
 /// Starts the connection.
 /// The connection will be alive until there is a channel error,
 /// protocol violation, or explicitly disconnected.
@@ -51,6 +50,7 @@ pub fn connect(
   clean_session: Bool,
   keep_alive_ms: Int,
   server_timeout_ms: Int,
+  auth: Option(packet.AuthOptions),
   will: Option(#(packet.MessageData, packet.QoS)),
 ) -> Result(Connection, String) {
   // Start linked, as this *really* shouldn't fail
@@ -60,7 +60,7 @@ pub fn connect(
       let messages = process.new_subject()
       let connect = process.new_subject()
       process.send(ack_subject, #(messages, connect))
-      establish_channel(messages, connect, clean_session, will)
+      establish_channel(messages, connect, clean_session, auth, will)
     })
 
   // Start monitoring and unlink
@@ -160,6 +160,7 @@ fn establish_channel(
   mailbox: Subject(Message),
   connect: Subject(Config),
   clean_session: Bool,
+  auth: Option(packet.AuthOptions),
   will: Option(#(packet.MessageData, packet.QoS)),
 ) -> Nil {
   // If we can't receive in time, the parent is very stuck, so dying is fine
@@ -187,13 +188,12 @@ fn establish_channel(
       connection_state,
     )
 
-  // TODO auth, will
   let connect_options =
     packet.ConnectOptions(
       clean_session:,
       client_id: config.client_id,
       keep_alive_seconds: config.keep_alive_ms / 1000,
-      auth: None,
+      auth:,
       will:,
     )
 

src/spoke/internal/packet.gleam

@@ -17,7 +17,7 @@ pub type MessageData {
 
 // MQTT does not allow using only a password
 pub type AuthOptions {
-  AuthOptions(user_name: String, password: Option(BitArray))
+  AuthOptions(username: String, password: Option(BitArray))
 }
 
 pub type ConnectOptions {

src/spoke/internal/packet/encode.gleam

@@ -60,7 +60,7 @@ pub fn connect(options: ConnectOptions) -> BytesTree {
     Some(auth) -> {
       let payload =
         payload
-        |> bytes_tree.append(string(auth.user_name))
+        |> bytes_tree.append(string(auth.username))
 
       case auth.password {
         Some(password) -> {

test/connect_test.gleam

@@ -140,6 +140,28 @@ pub fn connect_with_will_test() {
   spoke.disconnect(client)
 }
 
+pub fn connect_with_auth_test() {
+  let server = fake_server.start_server()
+  let client =
+    fake_server.default_options(server.port)
+    |> spoke.connect_with_id(default_client_id)
+    |> spoke.using_auth("user", Some(<<"Hunter2">>))
+    |> spoke.start
+
+  spoke.connect(client, True)
+
+  let server = fake_server.expect_connection_established(server)
+  fake_server.expect_packet_matching(server, fn(packet) {
+    let assert server_in.Connect(connect_data) = packet
+    case connect_data.auth {
+      Some(packet.AuthOptions("user", Some(<<"Hunter2">>))) -> True
+      _ -> False
+    }
+  })
+
+  spoke.disconnect(client)
+}
+
 fn connect_with_error(
   error: packet.ConnectError,
   client_id: String,

test/fake_server.gleam

@@ -1,6 +1,7 @@
 import gleam/bit_array
 import gleam/bytes_tree.{type BytesTree}
 import gleam/erlang/process
+import gleam/option.{None}
 import gleam/string
 import gleeunit/should
 import glisten/socket.{type ListenSocket, type Socket}
@@ -38,6 +39,7 @@ pub fn set_up_connected_client_with_timeout(
   let client =
     spoke.start_with_ms_keep_alive(
       "ping-client",
+      None,
       1000,
       timeout,
       default_options(server.port),

test/options_test.gleam

@@ -19,7 +19,7 @@ pub fn connect_options_test() {
   ) =
     spoke.default_tcp_options("")
     |> spoke.connect_with_id("client-id")
-    |> spoke.using_auth(spoke.AuthDetails("user", Some(<<"Hunter2">>)))
+    |> spoke.using_auth("user", Some(<<"Hunter2">>))
     |> spoke.keep_alive_seconds(42)
     |> spoke.server_timeout_ms(420)
     as "Connect option modifiers should be properly applied"

test/ping_test.gleam

@@ -1,5 +1,6 @@
 import fake_server.{type ConnectedServer}
 import gleam/erlang/process
+import gleam/option.{None}
 import spoke.{AtMostOnce}
 import spoke/internal/packet/server/incoming as server_in
 import spoke/internal/packet/server/outgoing as server_out
@@ -79,6 +80,7 @@ fn set_up_connected() -> #(spoke.Client, ConnectedServer) {
   let client =
     spoke.start_with_ms_keep_alive(
       "ping-client",
+      None,
       keep_alive,
       server_timeout,
       fake_server.default_options(server.port),