[x] Auth support

Procfile

@@ -1,4 +1,6 @@
+nsq_auth: MIX_ENV=test mix run --no-halt -e "NSQ.Test.AuthServer.start(6790)"
 nsqd1: nsqd --tcp-address=127.0.0.1:6750 --http-address=127.0.0.1:6751 --https-address=127.0.0.1:6752 --worker-id=512 --broadcast-address=127.0.0.1 --lookupd-tcp-address=127.0.0.1:6770 --lookupd-tcp-address=127.0.0.1:6780 --tls-required=false --tls-root-ca-file=test/ssl_keys/elixirNsq.pem --tls-key=test/ssl_keys/elixir_nsq.key --tls-cert=test/ssl_keys/elixir_nsq.crt
 nsqd2: nsqd --tcp-address=127.0.0.1:6760 --http-address=127.0.0.1:6761 --https-address=127.0.0.1:6762 --worker-id=513 --broadcast-address=127.0.0.1 --lookupd-tcp-address=127.0.0.1:6770 --lookupd-tcp-address=127.0.0.1:6780 --tls-required=false --tls-root-ca-file=test/ssl_keys/elixirNsq.pem --tls-key=test/ssl_keys/elixir_nsq.key --tls-cert=test/ssl_keys/elixir_nsq.crt
+nsqd3: nsqd --tcp-address=127.0.0.1:6765 --http-address=127.0.0.1:6766 --https-address=127.0.0.1:6767 --worker-id=514 --broadcast-address=127.0.0.1 --lookupd-tcp-address=127.0.0.1:6770 --lookupd-tcp-address=127.0.0.1:6780 --tls-required=false --tls-root-ca-file=test/ssl_keys/elixirNsq.pem --tls-key=test/ssl_keys/elixir_nsq.key --tls-cert=test/ssl_keys/elixir_nsq.crt --auth-http-address=127.0.0.1:6790
 nsqlookupd1: nsqlookupd --tcp-address=127.0.0.1:6770 --http-address=127.0.0.1:6771 --broadcast-address=127.0.0.1
 nsqlookupd2: nsqlookupd --tcp-address=127.0.0.1:6780 --http-address=127.0.0.1:6781 --broadcast-address=127.0.0.1

README.md

@@ -26,7 +26,7 @@ See these resources for more info on building client libraries:
 - [x] Include Procfile for running nsqd/nsqlookupd for tests
 - [x] Graceful connection closing
 - [x] TLS support
-- [ ] Auth support
+- [x] Auth support
 - [ ] Deflate support
 - [ ] Snappy support
 - [x] Delegates

lib/nsq/config.ex

@@ -107,7 +107,7 @@ defmodule NSQ.Config do
     msg_timeout: 60 * @seconds,
 
     # secret for nsqd authentication (requires nsqd 0.2.29+)
-    auth_secret: nil,
+    auth_secret: "",
 
     # function or module to deal with messages
     message_handler: nil,

lib/nsq/connection/initializer.ex

@@ -135,6 +135,13 @@ defmodule NSQ.Connection.Initializer do
       socket |> wait_for_ok(conn_state.config.read_timeout)
     end
 
+    if parsed["auth_required"] == true do
+      auth_cmd = encode({:auth, conn_state.config.auth_secret})
+      conn_state.socket |> Socket.Stream.send!(auth_cmd)
+      {:response, json} = recv_nsq_response(conn_state)
+      Logger.debug(json)
+    end
+
     {:ok, conn_state}
   end
 

lib/nsq/protocol.ex

@@ -15,6 +15,8 @@ defmodule NSQ.Protocol do
       {:identify, options} ->
         json = Poison.encode!(options)
         "IDENTIFY\n" <> <<byte_size(json) :: size(32)>> <> json
+      {:auth, secret_key} ->
+        "AUTH\n" <> <<byte_size(secret_key) :: size(32)>> <> secret_key
       {:pub, topic, data} ->
         "PUB #{topic}\n" <> << byte_size(data) :: size(32) >> <> data
       {:mpub, topic, data} ->

lib/nsq/test/auth_server.ex

@@ -0,0 +1,17 @@
+defmodule NSQ.Test.AuthServer do
+  # This sets up an auth server for NSQD (run from Procfile) so that we can
+  # test auth properly.
+  def start(port) do
+    [:ranch, :cowlib, :cowboy, :http_server] |> Enum.each(&Application.start/1)
+    HttpServer.start(path: "/auth", port: port, response: Poison.encode! %{
+      ttl: 3600,
+      identity: "johndoe",
+      identity_url: "http://127.0.0.1",
+      authorizations: [%{
+        permissions: ["subscribe", "publish"],
+        topic: ".*",
+        channels: [".*"]
+      }]
+    })
+  end
+end

mix.exs

@@ -36,6 +36,9 @@ defmodule ElixirNsq.Mixfile do
 
       # testing
       {:secure_random, "~> 0.2", only: :test},
+
+      # Small HTTP server for running tests
+      {:http_server, github: "MaxPower15/http_server", tag: "function-response", only: :test},
     ]
   end
 end

mix.lock

@@ -1,7 +1,11 @@
 %{"connection": {:hex, :connection, "1.0.1"},
+  "cowboy": {:hex, :cowboy, "1.0.4"},
+  "cowlib": {:hex, :cowlib, "1.0.2"},
+  "http_server": {:git, "https://github.com/MaxPower15/http_server.git", "a658e62b6dd7d99cb5d3287aa577a5b894de7f07", [tag: "function-response"]},
   "httpotion": {:hex, :httpotion, "2.1.0"},
   "ibrowse": {:git, "https://github.com/cmullaparthi/ibrowse.git", "ea3305d21f37eced4fac290f64b068e56df7de80", [tag: "v4.1.2"]},
   "poison": {:hex, :poison, "1.5.0"},
+  "ranch": {:hex, :ranch, "1.2.0"},
   "secure_random": {:hex, :secure_random, "0.2.0"},
   "socket": {:hex, :socket, "0.3.1"},
   "uuid": {:hex, :uuid, "1.1.2"}}

test/consumer_test.exs

@@ -603,4 +603,24 @@ defmodule NSQ.ConsumerTest do
     :timer.sleep 100
     assert NSQ.Consumer.starved?(consumer) == false
   end
+
+  test "auth" do
+    test_pid = self
+    {:ok, consumer} = NSQ.Consumer.Supervisor.start_link(@test_topic, @test_channel1, %NSQ.Config{
+      nsqds: [{"127.0.0.1", 6765}],
+      auth_secret: "abc",
+      message_handler: fn(body, msg) ->
+        assert body == "HTTP message"
+        assert msg.attempts == 1
+        send(test_pid, :handled)
+        :ok
+      end
+    })
+
+    HTTP.post("http://127.0.0.1:6766/put?topic=#{@test_topic}", [body: "HTTP message"])
+    assert_receive(:handled, 2000)
+
+    HTTP.post("http://127.0.0.1:6766/put?topic=#{@test_topic}", [body: "HTTP message"])
+    assert_receive(:handled, 2000)
+  end
 end