Add support for user authentication

.travis.yml

@@ -1,17 +1,16 @@
 language: elixir
+
 elixir:
-  - 1.1.1
-  - 1.2.1
+  - 1.3
+
 otp_release:
-  - 18.0
-  - 18.1
-sudo: required
-before_install:
-  - source /etc/lsb-release && echo "deb http://download.rethinkdb.com/apt $DISTRIB_CODENAME main" | sudo tee /etc/apt/sources.list.d/rethinkdb.list
-  - wget -qO- http://download.rethinkdb.com/apt/pubkey.gpg | sudo apt-key add -
-  - sudo apt-get update -qq
-  - sudo apt-get install rethinkdb -y --force-yes
-before_script: rethinkdb --daemon
+  - 18.3
+  - 19.1
+
 install:
+  - mix local.rebar --force
   - mix local.hex --force
   - mix deps.get --only test
+
+addons:
+    rethinkdb: '2.3'

lib/rethinkdb/connection.ex

@@ -174,15 +174,16 @@ defmodule RethinkDB.Connection do
 
   * `:host` - hostname to use to connect to database. Defaults to `'localhost'`.
   * `:port` - port on which to connect to database. Defaults to `28015`.
-  * `:auth_key` - authorization key to use with database. Defaults to `nil`.
+  * `:user` - user to use for authentication. Defaults to `"admin"`.
+  * `:pass` - password to use for authentication. Defaults to `""`.
   * `:db` - default database to use with queries. Defaults to `nil`.
   * `:sync_connect` - whether to have `init` block until a connection succeeds. Defaults to `false`.
   * `:max_pending` - Hard cap on number of concurrent requests. Defaults to `10000`
   * `:ssl` - a dict of options. Support SSL options:
       * `:ca_certs` - a list of file paths to cacerts.
   """
   def start_link(opts \\ []) do
-    args = Dict.take(opts, [:host, :port, :auth_key, :db, :sync_connect, :ssl, :max_pending])
+    args = Dict.take(opts, [:host, :port, :user, :pass, :db, :sync_connect, :ssl, :max_pending])
     Connection.start_link(__MODULE__, args, opts)
   end
 
@@ -195,7 +196,8 @@ defmodule RethinkDB.Connection do
     ssl = Dict.get(opts, :ssl)
     opts = Dict.put(opts, :host, host)
       |> Dict.put_new(:port, 28015)
-      |> Dict.put_new(:auth_key, "")
+      |> Dict.put_new(:user, "admin")
+      |> Dict.put_new(:pass, "")
       |> Dict.put_new(:max_pending, 10000)
       |> Dict.drop([:sync_connect])
       |> Enum.into(%{})
@@ -220,10 +222,10 @@ defmodule RethinkDB.Connection do
     end
   end
 
-  def connect(_info, state = %{config: %{host: host, port: port, auth_key: auth_key, transport: {transport, transport_opts}}}) do
+  def connect(_info, state = %{config: %{host: host, port: port, user: user, pass: pass, transport: {transport, transport_opts}}}) do
     case Transport.connect(transport, host, port, [active: false, mode: :binary] ++ transport_opts) do
       {:ok, socket} ->
-        case handshake(socket, auth_key) do
+        case handshake(socket, user, pass) do
           {:error, _} -> {:stop, :bad_handshake, state}
           :ok ->
             :ok = Transport.setopts(socket, [active: :once])
@@ -314,22 +316,91 @@ defmodule RethinkDB.Connection do
     :ok
   end
 
-  defp handshake(socket, auth_key) do
-    :ok = Transport.send(socket, << 0x400c2d20 :: little-size(32) >>)
-    :ok = Transport.send(socket, << :erlang.iolist_size(auth_key) :: little-size(32) >>)
-    :ok = Transport.send(socket, auth_key)
-    :ok = Transport.send(socket, << 0x7e6970c7 :: little-size(32) >>)
-    case recv_until_null(socket, "") do
-      "SUCCESS" -> :ok
-      error = {:error, _} -> error
+  defp handshake(socket, user, pass) do
+    # Sends the “magic number” for the protocol version.
+    case handshake_message(socket, << 0x34c2bdc3:: little-size(32) >>) do
+      {:ok, %{"success" => true}} ->
+        # Generates the client nonce.
+        client_nonce = :crypto.strong_rand_bytes(20)
+        |> Base.encode64
+
+        client_first_message = "n=#{user},r=#{client_nonce}"
+
+        scram = Poison.encode!(%{
+          protocol_version: 0,
+          authentication_method: "SCRAM-SHA-256",
+          authentication: "n,,#{client_first_message}"
+        })
+
+        # Sends the “client-first-message”
+        case handshake_message(socket, scram <> "\0") do
+          {:ok, %{"success" => true, "authentication" => server_first_message}} ->
+            auth = server_first_message
+            |> String.split(",")
+            |> Enum.map(&(String.split(&1, "=", parts: 2)))
+            |> Enum.into(%{}, &List.to_tuple/1)
+
+            # Verify server nonce.
+            server_nonce = auth["r"]
+            if String.starts_with?(server_nonce, client_nonce) do
+              iter = auth["i"]
+              |> String.to_integer
+
+              salt = auth["s"]
+              |> Base.decode64!
+
+              salted_pass = RethinkDB.Connection.PBKDF2.generate(pass, salt, iterations: iter)
+
+              client_final_message = "c=biws,r=#{server_nonce}"
+
+              auth_msg = Enum.join([
+                client_first_message,
+                server_first_message,
+                client_final_message
+              ], ",")
+
+              client_key = :crypto.hmac(:sha256, salted_pass, "Client Key")
+              server_key = :crypto.hmac(:sha256, salted_pass, "Server Key")
+              stored_key = :crypto.hash(:sha256, client_key)
+              client_sig = :crypto.hmac(:sha256, stored_key, auth_msg)
+              server_sig = :crypto.hmac(:sha256, server_key, auth_msg)
+
+              proof = :crypto.exor(client_key, client_sig)
+              |> Base.encode64
+
+              scram = Poison.encode!(%{authentication: "#{client_final_message},p=#{proof}"})
+
+              # Sends the “client-last-message”
+              case handshake_message(socket, scram <> "\0") do
+                {:ok, %{"success" => true, "authentication" => server_final_message}} ->
+                  auth = server_final_message
+                  |> String.split(",")
+                  |> Enum.map(&(String.split(&1, "=", parts: 2)))
+                  |> Enum.into(%{}, &List.to_tuple/1)
+
+                  # Verifies server signature.
+                  if server_sig == Base.decode64!(auth["v"]) do
+                    :ok
+                  else
+                    {:error, "Invalid server signature"}
+                  end
+                {:ok, %{"success" => false, "error" => reason}} ->
+                  {:error, reason}
+              end
+            else
+              {:error, "Invalid server nonce"}
+            end
+          {:ok, %{"success" => false, "error" => reason}} ->
+            {:error, reason}
+        end
     end
   end
 
-  defp recv_until_null(socket, acc) do
-    case Transport.recv(socket, 1) do
-      {:ok, "\0"} -> acc
-      {:ok, a}    -> recv_until_null(socket, acc <> a)
-      x = {:error, _} -> x
-    end
+  defp handshake_message(sock, data) do
+    with :ok 		<- Transport.send(sock, data),
+        {:ok, data} <- Transport.recv(sock, 0),
+    do: data
+		|> String.replace_suffix("\0", "")
+		|> Poison.decode
   end
 end

lib/rethinkdb/connection/pbkdf2.ex

@@ -0,0 +1,57 @@
+defmodule RethinkDB.Connection.PBKDF2 do
+  @moduledoc """
+  `PBKDF2` implements PBKDF2 (Password-Based Key Derivation Function 2),
+  part of PKCS #5 v2.0 (Password-Based Cryptography Specification).
+  It can be used to derive a number of keys for various purposes from a given
+  secret. This lets applications have a single secure secret, but avoid reusing
+  that key in multiple incompatible contexts.
+  see http://tools.ietf.org/html/rfc2898#section-5.2
+  """
+  use Bitwise
+
+  @max_length bsl(1, 32) - 1
+
+  @doc """
+  Returns a derived key suitable for use.
+  ## Options
+    * `:iterations` - defaults to 1000 (increase to at least 2^16 if used for passwords);
+    * `:length`     - a length in octets for the derived key. Defaults to 32;
+    * `:digest`     - an hmac function to use as the pseudo-random function. Defaults to `:sha256`;
+  """
+  def generate(secret, salt, opts \\ []) do
+    iterations = Keyword.get(opts, :iterations, 1000)
+    length = Keyword.get(opts, :length, 32)
+    digest = Keyword.get(opts, :digest, :sha256)
+
+    if length > @max_length do
+      raise ArgumentError, "length must be less than or equal to #{@max_length}"
+    else
+      generate(mac_fun(digest, secret), salt, iterations, length, 1, [], 0)
+    end
+  end
+
+  defp generate(_fun, _salt, _iterations, max_length, _block_index, acc, length)
+      when length >= max_length do
+    key = acc |> Enum.reverse |> IO.iodata_to_binary
+    <<bin::binary-size(max_length), _::binary>> = key
+    bin
+  end
+
+  defp generate(fun, salt, iterations, max_length, block_index, acc, length) do
+    initial = fun.(<<salt::binary, block_index::integer-size(32)>>)
+    block   = iterate(fun, iterations - 1, initial, initial)
+    generate(fun, salt, iterations, max_length, block_index + 1,
+             [block | acc], byte_size(block) + length)
+  end
+
+  defp iterate(_fun, 0, _prev, acc), do: acc
+
+  defp iterate(fun, iteration, prev, acc) do
+    next = fun.(prev)
+    iterate(fun, iteration - 1, next, :crypto.exor(next, acc))
+  end
+
+  defp mac_fun(digest, secret) do
+    &:crypto.hmac(digest, secret, &1)
+  end
+end

test/connection_test.exs

@@ -88,6 +88,25 @@ defmodule ConnectionTest do
     assert data == ["new_test_table"]
   end
 
+  test "connection authenticates with admin" do
+    {:ok, c} = RethinkDB.Connection.start_link(user: "admin", pass: "")
+    {:ok, _} =  table_list() |> RethinkDB.run(c)
+  end
+
+  test "connection fails to authenticate with invalid user" do
+    {:ok, c} = RethinkDB.Connection.start_link(user: "bob", pass: "bobpwd")
+    Process.flag(:trap_exit, true)
+    assert {:bad_handshake, _} = catch_exit(table_list() |> RethinkDB.run(c))
+  end
+
+  test "connection authenticates with newly created user" do
+    {:ok, c} = RethinkDB.Connection.start_link()
+    assert {:ok, %{data: %{"inserted" => 1}}} = db("rethinkdb") |> table("users") |> insert(%{id: "bob", password: "bobpwd"}) |> RethinkDB.run(c)
+    {:ok, d} = RethinkDB.Connection.start_link(user: "bob", pass: "bobpwd")
+    assert {:ok, _} = table_list() |> RethinkDB.run(d)
+    assert {:ok, %{data: %{"deleted"  => 1}}} =  db("rethinkdb") |> table("users") |> get("bob") |> delete() |> RethinkDB.run(c)
+  end
+
   test "connection accepts max_pending" do
     {:ok, c} = RethinkDB.Connection.start_link(max_pending: 1)
     res = Enum.map(1..100, fn (_) ->
@@ -149,7 +168,7 @@ defmodule ConnectionRunTest do
 
   test "run with :noreply option" do
     :ok = make_array([1,2,3]) |> run(noreply: true)
-    noreply_wait 
+    noreply_wait
   end
 
   test "run with :profile options" do