introduce auth_chain, abstract authenticator and the basic authenticator

src/lavinmq/amqp/connection_factory.cr

@@ -8,14 +8,14 @@ module LavinMQ
     class ConnectionFactory < LavinMQ::ConnectionFactory
       Log = LavinMQ::Log.for "amqp.connection_factory"
 
-      def start(socket, connection_info, vhosts, users) : Client?
+      def start(socket, connection_info, vhosts, users, auth_chain) : Client?
         remote_address = connection_info.src
         socket.read_timeout = 15.seconds
         metadata = ::Log::Metadata.build({address: remote_address.to_s})
         logger = Logger.new(Log, metadata)
         if confirm_header(socket, logger)
           if start_ok = start(socket, logger)
-            if user = authenticate(socket, remote_address, users, start_ok, logger)
+            if user = authenticate(socket, remote_address, users, start_ok, logger, auth_chain)
               if tune_ok = tune(socket, logger)
                 if vhost = open(socket, vhosts, user, logger)
                   socket.read_timeout = heartbeat_timeout(tune_ok)
@@ -100,10 +100,10 @@ module LavinMQ
         end
       end
 
-      def authenticate(socket, remote_address, users, start_ok, log)
+      def authenticate(socket, remote_address, users, start_ok, log, auth_chain)
         username, password = credentials(start_ok)
         user = users[username]?
-        return user if user && user.password && user.password.not_nil!.verify(password) &&
+        return user if user && auth_chain.authenticate(username, password) &&
                        guest_only_loopback?(remote_address, user)
 
         if user.nil?

src/lavinmq/auth/authenticator.cr

@@ -0,0 +1,24 @@
+module LavinMQ
+  module Auth
+    abstract class Authenticator
+      Log = LavinMQ::Log.for "auth.handler"
+      @log = Logger.new(Log)
+      property successor : Authenticator?
+
+      abstract def authenticate(username : String, password : String)
+
+      def set_successor(service : Authenticator) : Authenticator
+        @successor = service
+        service
+      end
+
+      def try_next(username : String, password : String)
+        if successor = @successor
+          successor.authenticate(username, password)
+        else
+          nil
+        end
+      end
+    end
+  end
+end

src/lavinmq/auth/authenticators/basic.cr

@@ -0,0 +1,18 @@
+require "../authenticator"
+require "../../server"
+
+module LavinMQ
+  module Auth
+    class BasicAuthenticator < LavinMQ::Auth::Authenticator
+      def initialize(@users : UserStore)
+      end
+
+      def authenticate(username : String, password : String)
+        user = @users[username]
+        return user if user && user.password && user.password.not_nil!.verify(password)
+        @log.info { "Basic authentication failed" }
+        try_next(username, password)
+      end
+    end
+  end
+end

src/lavinmq/auth/chain.cr

@@ -0,0 +1,43 @@
+require "./authenticator"
+require "./authenticators/basic"
+
+module LavinMQ
+  module Auth
+    class Chain
+      @first : Authenticator?
+
+      def initialize(users : UserStore)
+        backends = Config.instance.auth_backends
+        if backends.nil? || backends.size == 0
+          add_handler(BasicAuthenticator.new(users))
+        else
+          backends.each do |backend|
+            case backend
+            when "basic"
+              add_handler(BasicAuthenticator.new(users))
+            else
+              raise "Unsupported authentication backend: #{backend}"
+            end
+          end
+        end
+      end
+
+      def add_handler(auth : Authenticator)
+        if first = @first
+          current = first
+          while successor = current.@successor
+            current = successor
+          end
+          current.set_successor(auth)
+        else
+          @first = auth
+        end
+        self
+      end
+
+      def authenticate(username : String, password : String)
+        @first.try &.authenticate(username, password)
+      end
+    end
+  end
+end

src/lavinmq/config.cr

@@ -62,6 +62,7 @@ module LavinMQ
     property default_consumer_prefetch = UInt16::MAX
     property yield_each_received_bytes = 131_072    # max number of bytes to read from a client connection without letting other tasks in the server do any work
     property yield_each_delivered_bytes = 1_048_576 # max number of bytes sent to a client without tending to other tasks in the server
+    property auth_backends : Array(String)? = nil
     @@instance : Config = self.new
 
     def self.instance : LavinMQ::Config

src/lavinmq/server.cr

@@ -16,6 +16,7 @@ require "./client/client"
 require "./client/connection_factory"
 require "./amqp/connection_factory"
 require "./stats"
+require "./auth/chain"
 
 module LavinMQ
   class Server
@@ -38,6 +39,7 @@ module LavinMQ
       @vhosts = VHostStore.new(@data_dir, @users, @replicator)
       @parameters = ParameterStore(Parameter).new(@data_dir, "parameters.json", @replicator)
       @amqp_connection_factory = LavinMQ::AMQP::ConnectionFactory.new
+      @auth_chain = LavinMQ::Auth::Chain.new(@users)
       apply_parameter
       spawn stats_loop, name: "Server#stats_loop"
     end
@@ -245,7 +247,7 @@ module LavinMQ
     end
 
     def handle_connection(socket, connection_info)
-      client = @amqp_connection_factory.start(socket, connection_info, @vhosts, @users)
+      client = @amqp_connection_factory.start(socket, connection_info, @vhosts, @users, @auth_chain)
     ensure
       socket.close if client.nil?
     end