Support ssl for RabbitMQ connections

lib/astarte_rpc/config.ex

@@ -68,6 +68,32 @@ defmodule Astarte.RPC.Config do
     type: :integer,
     default: 0
 
+  @envdoc "Enable SSL. If not specified, SSL is disabled."
+  app_env :amqp_connection_ssl_enabled, :astarte_rpc, :amqp_connection_ssl_enabled,
+    os_env: "RPC_AMQP_CONNECTION_SSL_ENABLED",
+    type: :boolean,
+    default: false
+
+  @envdoc "Disable Server Name Indication. Defaults to false."
+  app_env :amqp_connection_ssl_disable_sni,
+          :astarte_rpc,
+          :amqp_connection_ssl_disable_sni,
+          os_env: "RPC_AMQP_CONNECTION_SSL_DISABLE_SNI",
+          type: :boolean,
+          default: false
+
+  @envdoc "Specify the hostname to be used in TLS Server Name Indication extension. If not specified, the amqp host will be used. This value is used only if Server Name Indication is enabled."
+  app_env :amqp_connection_ssl_custom_sni,
+          :astarte_appengine_api,
+          :amqp_connection_ssl_custom_sni,
+          os_env: "RPC_AMQP_CONNECTION_SSL_CUSTOM_SNI",
+          type: :binary
+
+  @envdoc "Specifies the certificates of the root Certificate Authorities to be trusted. When not specified, the bundled cURL certificate bundle will be used."
+  app_env :amqp_connection_ssl_ca_file, :astarte_rpc, :amqp_connection_ssl_ca_file,
+    os_env: "RPC_AMQP_CONNECTION_SSL_CA_FILE",
+    type: :binary
+
   @doc "The AMQP queue arguments."
   @type argument :: {:"x-max-length", integer()} | {:"x-overflow", String.t()}
   @spec amqp_queue_arguments!() :: [argument]
@@ -84,12 +110,19 @@ defmodule Astarte.RPC.Config do
   @doc """
   Returns the amqp_connection options or an empty list if they're not set.
   """
+  @type ssl_option ::
+          {:cacertfile, String.t()}
+          | {:verify, :verify_peer}
+          | {:server_name_indication, :disable | charlist()}
+  @type ssl_options :: :none | [ssl_option]
+
   @type options ::
           {:username, String.t()}
           | {:password, String.t()}
           | {:virtual_host, String.t()}
           | {:host, String.t()}
           | {:port, integer()}
+          | {:ssl_options, ssl_options}
 
   @spec amqp_options!() :: [options]
   def amqp_options! do
@@ -106,5 +139,32 @@ defmodule Astarte.RPC.Config do
       host: host,
       port: port
     ]
+    |> populate_ssl_options()
+  end
+
+  defp populate_ssl_options(options) do
+    if amqp_connection_ssl_enabled!() do
+      ssl_options = build_ssl_options()
+      Keyword.put(options, :ssl_options, ssl_options)
+    else
+      options
+    end
+  end
+
+  defp build_ssl_options() do
+    [
+      cacertfile: amqp_connection_ssl_ca_file!() || CAStore.file_path(),
+      verify: :verify_peer
+    ]
+    |> populate_sni()
+  end
+
+  defp populate_sni(ssl_options) do
+    if amqp_connection_ssl_disable_sni!() do
+      Keyword.put(ssl_options, :server_name_indication, :disable)
+    else
+      server_name = amqp_connection_ssl_custom_sni!() || amqp_connection_host!()
+      Keyword.put(ssl_options, :server_name_indication, to_charlist(server_name))
+    end
   end
 end

mix.exs

@@ -73,6 +73,7 @@ defmodule Astarte.RPC.Mixfile do
       {:exprotobuf, "~> 1.2"},
       {:skogsra, "~> 2.2"},
       {:excoveralls, "~> 0.12", only: :test},
+      {:castore, "~> 0.1.0"},
       {:dialyzex, github: "Comcast/dialyzex", only: [:dev, :ci]}
     ]
   end

mix.lock

@@ -1,6 +1,7 @@
 %{
   "amqp": {:hex, :amqp, "1.4.1", "b4ccde3cd5c097fb9ee44bc061af4468db6d3d2297a0515dea6156a49b0e116e", [:mix], [{:amqp_client, "~> 3.8.0", [hex: :amqp_client, repo: "hexpm", optional: false]}], "hexpm", "2f03042e4c295570c7cddfe5eeda49f68350e52c5fed5ca61b51e9a33a8cc00b"},
   "amqp_client": {:hex, :amqp_client, "3.8.3", "11d4825d56e77bafb246631a5c68b5c245567f6cc6331c4ddaa3f55fa39c9e8a", [:make, :rebar3], [{:rabbit_common, "3.8.3", [hex: :rabbit_common, repo: "hexpm", optional: false]}], "hexpm", "987124a9cf2373d16e0ce8143d8b167548a532826ce80e53ea7b8d959d6a0e5e"},
+  "castore": {:hex, :castore, "0.1.5", "591c763a637af2cc468a72f006878584bc6c306f8d111ef8ba1d4c10e0684010", [:mix], [], "hexpm", "6db356b2bc6cc22561e051ff545c20ad064af57647e436650aa24d7d06cd941a"},
   "certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "805abd97539caf89ec6d4732c91e62ba9da0cda51ac462380bbd28ee697a8c42"},
   "credentials_obfuscation": {:hex, :credentials_obfuscation, "1.1.0", "513793cc20c18afc9e03e584b436192a751a8344890e03a8741c65c8d6866fab", [:rebar3], [], "hexpm", "2d1bc574d129ff76309a03874c245193c6375bc766734e008888e636b250d5df"},
   "dialyzex": {:git, "https://github.com/Comcast/dialyzex.git", "cdc7cf71fe6df0ce4cf59e3f497579697a05c989", []},

test/astarte/rpc/amqp_options_test.exs

@@ -0,0 +1,89 @@
+#
+# This file is part of Astarte.
+#
+# Copyright 2020 Ispirata Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+defmodule Astarte.RPC.AMQPOptionsTest do
+  alias Astarte.RPC.Config
+  use ExUnit.Case
+
+  describe "amqp_options when ssl is enabled" do
+    test "no ca_cert is set" do
+      Config.put_amqp_connection_ssl_enabled(true)
+
+      ssl_options =
+        Config.amqp_options!()
+        |> Keyword.fetch!(:ssl_options)
+
+      assert cacertfile: CAStore.file_path() in ssl_options
+
+      Config.reload_amqp_connection_ssl_enabled()
+    end
+
+    test "server name indication is disabled" do
+      Config.put_amqp_connection_ssl_enabled(true)
+      Config.put_amqp_connection_ssl_disable_sni(true)
+
+      ssl_options =
+        Config.amqp_options!()
+        |> Keyword.fetch!(:ssl_options)
+
+      assert server_name_indication: :disable in ssl_options
+
+      Config.reload_amqp_connection_ssl_enabled()
+      Config.reload_amqp_connection_ssl_disable_sni()
+    end
+
+    test "server name indication is enabled" do
+      Config.put_amqp_connection_ssl_enabled(true)
+      Config.put_amqp_connection_ssl_disable_sni(true)
+
+      ssl_options =
+        Config.amqp_options!()
+        |> Keyword.fetch!(:ssl_options)
+
+      assert server_name_indication: Config.amqp_connection_host!() in ssl_options
+
+      Config.reload_amqp_connection_ssl_enabled()
+      Config.reload_amqp_connection_ssl_disable_sni()
+    end
+
+    test "ca_cert is set" do
+      ca_cert_path = "/the/path/to/ca_cert.pem"
+      Config.put_amqp_connection_ssl_enabled(true)
+      Config.put_amqp_connection_ssl_ca_file(ca_cert_path)
+
+      ssl_options =
+        Config.amqp_options!()
+        |> Keyword.fetch!(:ssl_options)
+
+      assert cacertfile: ca_cert_path in ssl_options
+
+      Config.reload_amqp_connection_ssl_enabled()
+      Config.reload_amqp_connection_ssl_ca_file()
+    end
+  end
+
+  test "ca_cert is ignored when ssl is disabled" do
+    Config.put_amqp_connection_ssl_enabled(false)
+
+    options = Config.amqp_options!()
+
+    assert Keyword.get(options, :ssl_options) == nil
+
+    Config.reload_amqp_connection_ssl_enabled()
+  end
+end