Issue 246 auth backend

spec/auth_spec.cr

@@ -0,0 +1,62 @@
+require "./spec_helper"
+
+class MockAuthService < LavinMQ::AuthenticationService
+  property? should_succeed : Bool
+  property last_username : String?
+  property last_password : String?
+
+  def initialize(@should_succeed = false)
+  end
+
+  def authorize?(username : String, password : String) : Bool
+    @last_username = username
+    @last_password = password
+
+    if @should_succeed
+      true
+    else
+      try_next(username, password)
+    end
+  end
+end
+
+describe LavinMQ::AuthenticationChain do
+  describe "#authorize?" do
+    it "returns nil when no services are configured" do
+      chain = LavinMQ::AuthenticationChain.new
+      chain.authorize?("user", "pass").should be_nil
+    end
+
+    it "tries services in order until success" do
+      # Arrange
+      chain = LavinMQ::AuthenticationChain.new
+      service1 = MockAuthService.new(should_succeed: false)
+      service2 = MockAuthService.new(should_succeed: true)
+      service3 = MockAuthService.new(should_succeed: true)
+
+      chain.add_service(service1)
+      chain.add_service(service2)
+      chain.add_service(service3)
+
+      # Act
+      user = chain.authorize?("test_user", "test_pass")
+
+      # Assert
+      user.should_not be_nil
+      service1.last_username.should eq("test_user")
+      service2.last_username.should eq("test_user")
+      service3.last_username.should be_nil # Ne devrait pas être appelé
+    end
+
+    it "returns nil if all services fail" do
+      chain = LavinMQ::AuthenticationChain.new
+      service1 = MockAuthService.new(should_succeed: false)
+      service2 = MockAuthService.new(should_succeed: false)
+
+      chain.add_service(service1)
+      chain.add_service(service2)
+
+      chain.authorize?("user", "pass").should be_false
+    end
+  end
+end

spec/cache_spec.cr

@@ -0,0 +1,49 @@
+require "./spec_helper"
+
+describe LavinMQ::Cache do
+  cache = LavinMQ::Cache(String, String).new(1.seconds)
+
+  it "should set key" do
+    cache.set("key1", "allow").should eq "allow"
+  end
+
+  it "should get exitant key" do
+    cache.set("keyget", "deny")
+    cache.get?("keyget").should eq "deny"
+  end
+
+  it "should invalid cache after 1 second" do
+    cache.set("keyinvalid", "expired")
+    sleep(2.seconds)
+    cache.get?("keyinvalid").should be_nil
+  end
+
+  it "shoudl delete key" do
+    cache.set("keydelete", "deleted")
+    cache.delete("keydelete")
+    cache.get?("keydelete").should be_nil
+  end
+
+  it "should cleanup expired entry" do
+    cache.set("clean1", "expired1")
+    cache.set("clean2", "expired2")
+    cache.set("clean3", "valid", 10.seconds)
+    sleep(2.seconds)
+    cache.get?("clean1").should be_nil
+    cache.get?("clean2").should be_nil
+    cache.get?("clean3").should eq "valid"
+  end
+
+  it "should fetch data if key exists and not expired" do
+      result = cache.fetch("key") do
+        "fetchvalue"
+      end
+      result.should eq("fetchvalue")
+
+      result = cache.fetch("key") do
+        "fake fetchvalue"
+      end
+      result.should eq("fetchvalue")
+  end
+
+end

spec/config_spec.cr

@@ -10,6 +10,7 @@ describe LavinMQ::Config do
         data_dir = /tmp/lavinmq-spec
         [mgmt]
         [amqp]
+        [auth]
       CONFIG
     end
     config = LavinMQ::Config.new

spec/spec_helper.cr

@@ -14,6 +14,8 @@ require "../src/lavinmq/server"
 require "../src/lavinmq/http/http_server"
 require "http/client"
 require "amqp-client"
+require "http/server"
+require "json"
 
 LavinMQ::Config.instance.data_dir = "/tmp/lavinmq-spec"
 LavinMQ::Config.instance.segment_size = 512 * 1024
@@ -106,6 +108,159 @@ def with_http_server(&)
   end
 end
 
+class AuthBackend
+  getter? running : Bool
+
+  # Structure pour représenter une requête d'authentification
+  struct AuthRequest
+    include JSON::Serializable
+    property username : String
+    property password : String
+  end
+
+  struct AuthVhostRequest
+    include JSON::Serializable
+    property username : String
+    property vhost : String
+    property action : String
+  end
+
+  struct AuthResourceRequest
+    include JSON::Serializable
+    property username : String
+    property vhost : String
+    property resource : String
+    property name : String
+    property permission : String
+  end
+
+  # Simule une base de données d'utilisateurs
+  USERS = {
+    "admin" => "secret",
+    "user1" => "password123",
+  }
+
+  # Simule des permissions sur les vhosts
+  VHOST_PERMISSIONS = {
+    "admin" => ["vhost1", "vhost2"],
+    "user1" => ["vhost1"],
+  }
+
+  # Simule des permissions sur les ressources
+  RESOURCE_PERMISSIONS = {
+    "admin" => {"exchange1" => "configure", "queue1" => "write"},
+    "user1" => {"queue1" => "read"},
+  }
+
+  def initialize(@address : String = "0.0.0.0", @port : Int32 = 8081)
+    @server = HTTP::Server.new do |context|
+      handle_request(context)
+    end
+    @running = false
+  end
+
+  def run
+    if @running
+      puts "Server is already running at http://#{@address}:#{@port}"
+      return
+    end
+
+    @running = true
+    spawn do
+      puts "Starting server at http://#{@address}:#{@port}"
+      @server.listen(@address, @port)
+    rescue ex : Exception
+      puts "Server encountered an error: #{ex.message}"
+      @running = false
+    end
+  end
+
+  def kill
+    if !@running
+      puts "Server is not running."
+      return
+    end
+
+    puts "Stopping server..."
+    @server.close
+    @running = false
+    puts "Server stopped."
+  end
+
+  private def handle_request(context : HTTP::Server::Context)
+    request = context.request
+    response = context.response
+
+    case request.path
+    when "/auth/user"
+      handle_user_auth(request, response)
+    when "/auth/vhost"
+      handle_vhost_auth(request, response)
+    when "/auth/resource"
+      handle_resource_auth(request, response)
+    else
+      response.status_code = 404
+      response.print("Not Found")
+    end
+  end
+
+  private def handle_user_auth(request : HTTP::Request, response : HTTP::Server::Response)
+    if body = request.body
+      auth_request = AuthRequest.from_json(body)
+
+      if USERS[auth_request.username] == auth_request.password
+        allow_response(response)
+      else
+        deny_response(response)
+      end
+    else
+      deny_response(response)
+    end
+  end
+
+  private def handle_vhost_auth(request : HTTP::Request, response : HTTP::Server::Response)
+    if body = request.body
+      vhost_request = AuthVhostRequest.from_json(body)
+
+      if VHOST_PERMISSIONS[vhost_request.username].includes?(vhost_request.vhost)
+        allow_response(response)
+      else
+        deny_response(response)
+      end
+    else
+      deny_response(response)
+    end
+  end
+
+  private def handle_resource_auth(request : HTTP::Request, response : HTTP::Server::Response)
+    if body = request.body
+      resource_request = AuthResourceRequest.from_json(body)
+
+      if user_perms = RESOURCE_PERMISSIONS[resource_request.username]?
+        if user_perms[resource_request.name] == resource_request.permission
+          allow_response(response)
+        else
+          deny_response(response)
+        end
+      else
+        deny_response(response)
+      end
+    else
+      deny_response(response)
+    end
+  end
+
+  private def allow_response(response)
+    response.status_code = 200
+    response.print "allow"
+  end
+
+  private def deny_response(response)
+    response.status_code = 403
+    response.print "deny"
+  end
+end
+
 struct HTTPSpecHelper
   def initialize(@addr : String)
   end

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,11 @@ 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)
+        # TODO: resolve how other provider can produce LavinMQ::User.
         user = users[username]?
-        return user if user && user.password && user.password.not_nil!.verify(password) &&
+        return user if user && auth_chain.authorize?(username, password) &&
                        guest_only_loopback?(remote_address, user)
 
         if user.nil?

src/lavinmq/auth/auth_chain.cr

@@ -0,0 +1,68 @@
+require "./services/auth_service"
+require "./services/local_auth_service"
+require "./services/http_service"
+require "../cache"
+require "../config"
+
+module LavinMQ
+  class AuthenticationChain
+    @first_service : AuthenticationService?
+    @auth_cache : Cache(String, Bool)
+
+    def initialize(@users_store : UserStore, @auth_cache = Cache(String, Bool).new(Config.instance.auth_cache_time))
+      @first_service = nil
+
+      backends = Config.instance.auth_backends
+
+      if backends.empty?
+        add_service(LocalAuthService.new(@users_store))
+      else
+        sorted_backends = backends.to_a.sort_by(&.first)
+
+        sorted_backends.each do |(_, backend_type)|
+          case backend_type
+          when "http"
+            # TODO: Verify config before init service
+            add_service(HttpAuthService.new(
+              Config.instance.auth_http_method.not_nil!,
+              Config.instance.auth_http_user_path.not_nil!,
+              Config.instance.auth_http_vhost_path.not_nil!,
+              Config.instance.auth_http_resource_path.not_nil!,
+              Config.instance.auth_http_topic_path.not_nil!
+            ))
+          when "internal"
+            add_service(LocalAuthService.new(@users_store))
+          else
+            puts "Unknwon backend"
+          end
+        end
+
+        unless backends.values.includes?("internal")
+          add_service(LocalAuthService.new(@users_store))
+        end
+      end
+    end
+
+    def add_service(service : AuthenticationService)
+      if first = @first_service
+        current = first
+        while next_service = current.next_service
+          current = next_service
+        end
+        current.then(service)
+      else
+        @first_service = service
+      end
+      self
+    end
+
+    def authorize?(username : String, password : String)
+      if value = @auth_cache.get?(username)
+        return value
+      end
+      if service = @first_service
+        @auth_cache.set(username, service.authorize?(username, password))
+      end
+    end
+  end
+end