Merge pull request #1068 from Annopaolo/reliable-public-key-retrieval

Backends: fix corner case error in public key retrieval

Author: matt-mazzucato

CHANGELOG.md

@@ -22,6 +22,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - [astarte_pairing] Correctly handle Cassandra `varchar`s.
 - [astarte_realm_management] Correctly handle Cassandra `varchar`s.
 - [astarte_trigger_engine] Correctly handle Cassandra `varchar`s.
+- [astarte_pairing] Fix a corner case in the realm public key retrieval
+  when connection to the database might fail.
+- [astarte_realm_management] Fix a corner case in the realm public key retrieval
+  when connection to the database might fail.
+- [astarte_appengine_api] Fix a corner case in the realm public key retrieval
+  when connection to the database might fail.
 
 ## [1.2.0] - 2024-07-02
 ### Fixed

apps/astarte_appengine_api/lib/astarte_appengine_api/auth/auth.ex

@@ -18,23 +18,14 @@
 
 defmodule Astarte.AppEngine.API.Auth do
   alias Astarte.AppEngine.API.Queries
-  alias Astarte.DataAccess.Database
+  alias Astarte.AppEngine.API.Config
+  alias Astarte.Core.CQLUtils
 
   require Logger
 
   def fetch_public_key(realm) do
-    with {:ok, client} <- Database.connect(realm: realm),
-         {:ok, public_key} <- Queries.fetch_public_key(client) do
-      {:ok, public_key}
-    else
-      {:error, :public_key_not_found} ->
-        _ = Logger.warning("No public key found in realm #{realm}.", tag: "no_public_key_found")
-        {:error, :public_key_not_found}
+    keyspace = CQLUtils.realm_name_to_keyspace_name(realm, Config.astarte_instance_id!())
 
-      {:error, :database_connection_error} ->
-        _ = Logger.info("Auth request for unexisting realm #{realm}.", tag: "unexisting_realm")
-        # TODO: random busy wait here to prevent realm enumeration
-        {:error, :not_existing_realm}
-    end
+    Queries.fetch_public_key(keyspace)
   end
 end

apps/astarte_appengine_api/lib/astarte_appengine_api/queries.ex

@@ -23,23 +23,68 @@ defmodule Astarte.AppEngine.API.Queries do
 
   require Logger
 
-  def fetch_public_key(client) do
-    query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(
-        "SELECT blobAsVarchar(value) FROM kv_store WHERE group='auth' AND key='jwt_public_key_pem'"
-      )
+  @keyspace_does_not_exist_regex ~r/Keyspace (.*) does not exist/
+
+  def fetch_public_key(keyspace_name) do
+    case Xandra.Cluster.run(:xandra, &do_fetch_public_key(keyspace_name, &1)) do
+      {:ok, pem} ->
+        {:ok, pem}
+
+      {:error, %Xandra.ConnectionError{} = err} ->
+        _ =
+          Logger.warning("Database connection error #{Exception.message(err)}.",
+            tag: "database_connection_error"
+          )
 
-    result =
-      DatabaseQuery.call!(client, query)
-      |> DatabaseResult.head()
+        {:error, :database_connection_error}
+
+      {:error, %Xandra.Error{} = err} ->
+        handle_xandra_error(err)
+    end
+  end
+
+  defp do_fetch_public_key(keyspace_name, conn) do
+    query = """
+    SELECT blobAsVarchar(value)
+    FROM #{keyspace_name}.kv_store
+    WHERE group='auth' AND key='jwt_public_key_pem';
+    """
 
-    case result do
-      ["system.blobasvarchar(value)": public_key] ->
-        {:ok, public_key}
+    with {:ok, prepared} <- Xandra.prepare(conn, query),
+         {:ok, page} <-
+           Xandra.execute(conn, prepared, %{},
+             uuid_format: :binary,
+             consistency: :quorum
+           ) do
+      case Enum.to_list(page) do
+        [%{"system.blobasvarchar(value)" => pem}] ->
+          {:ok, pem}
+
+        [] ->
+          {:error, :public_key_not_found}
+      end
+    end
+  end
+
+  defp handle_xandra_error(error) do
+    %Xandra.Error{message: message} = error
+
+    case Regex.run(@keyspace_does_not_exist_regex, message) do
+      [_message, keyspace] ->
+        Logger.warning("Keyspace #{keyspace} does not exist.",
+          tag: "realm_not_found"
+        )
+
+        {:error, :not_existing_realm}
+
+      nil ->
+        _ =
+          Logger.warning(
+            "Database error, cannot get realm public key: #{Exception.message(error)}.",
+            tag: "database_error"
+          )
 
-      :empty_dataset ->
-        {:error, :public_key_not_found}
+        {:error, :database_error}
     end
   end
 

apps/astarte_pairing/lib/astarte_pairing/engine.ex

@@ -56,27 +56,10 @@ defmodule Astarte.Pairing.Engine do
     end
   end
 
-  def get_agent_public_key_pems(realm) do
-    keyspace = CQLUtils.realm_name_to_keyspace_name(realm, Config.astarte_instance_id!())
+  def get_agent_public_key_pems(realm_name) do
+    keyspace = CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
 
-    cqex_options =
-      Config.cqex_options!()
-      |> Keyword.put(:keyspace, keyspace)
-
-    with {:ok, client} <-
-           Client.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ),
-         {:ok, jwt_pems} <- Queries.get_agent_public_key_pems(client) do
-      {:ok, jwt_pems}
-    else
-      {:error, :shutdown} ->
-        {:error, :realm_not_found}
-
-      {:error, reason} ->
-        {:error, reason}
-    end
+    Queries.get_agent_public_key_pems(keyspace)
   end
 
   def get_credentials(

apps/astarte_pairing/lib/astarte_pairing/queries.ex

@@ -28,29 +28,68 @@ defmodule Astarte.Pairing.Queries do
   require Logger
 
   @protocol_revision 1
+  @keyspace_does_not_exist_regex ~r/Keyspace (.*) does not exist/
 
-  def get_agent_public_key_pems(client) do
-    get_jwt_public_key_pem = """
+  def get_agent_public_key_pems(keyspace_name) do
+    case Xandra.Cluster.run(:xandra, &do_get_agent_public_key_pems(keyspace_name, &1)) do
+      {:ok, pems} ->
+        {:ok, pems}
+
+      {:error, %Xandra.ConnectionError{} = err} ->
+        _ =
+          Logger.warning("Database connection error #{Exception.message(err)}.",
+            tag: "database_connection_error"
+          )
+
+        {:error, :database_connection_error}
+
+      {:error, %Xandra.Error{} = err} ->
+        handle_xandra_error(err)
+    end
+  end
+
+  defp handle_xandra_error(error) do
+    %Xandra.Error{message: message} = error
+
+    case Regex.run(@keyspace_does_not_exist_regex, message) do
+      [_message, keyspace] ->
+        Logger.warning("Keyspace #{keyspace} does not exist.",
+          tag: "realm_not_found"
+        )
+
+        {:error, :realm_not_found}
+
+      nil ->
+        _ =
+          Logger.warning(
+            "Database error, cannot get realm public key: #{Exception.message(error)}.",
+            tag: "database_error"
+          )
+
+        {:error, :database_error}
+    end
+  end
+
+  defp do_get_agent_public_key_pems(keyspace_name, conn) do
+    query = """
     SELECT blobAsVarchar(value)
-    FROM kv_store
+    FROM #{keyspace_name}.kv_store
     WHERE group='auth' AND key='jwt_public_key_pem';
     """
 
-    # TODO: add additional keys
-    query =
-      Query.new()
-      |> Query.statement(get_jwt_public_key_pem)
-
-    with {:ok, res} <- Query.call(client, query),
-         ["system.blobasvarchar(value)": pem] <- Result.head(res) do
-      {:ok, [pem]}
-    else
-      :empty_dataset ->
-        {:error, :public_key_not_found}
+    with {:ok, prepared} <- Xandra.prepare(conn, query),
+         {:ok, page} <-
+           Xandra.execute(conn, prepared, %{},
+             uuid_format: :binary,
+             consistency: :quorum
+           ) do
+      case Enum.to_list(page) do
+        [%{"system.blobasvarchar(value)" => pem}] ->
+          {:ok, [pem]}
 
-      error ->
-        Logger.warning("DB error: #{inspect(error)}")
-        {:error, :database_error}
+        [] ->
+          {:error, :public_key_not_found}
+      end
     end
   end
 

apps/astarte_realm_management/lib/astarte_realm_management/engine.ex

@@ -477,20 +477,7 @@ defmodule Astarte.RealmManagement.Engine do
     keyspace_name =
       CQLUtils.realm_name_to_keyspace_name(realm_name, Config.astarte_instance_id!())
 
-    cqex_options =
-      Config.cqex_options!()
-      |> Keyword.put(:keyspace, keyspace_name)
-
-    with {:ok, client} <-
-           DatabaseClient.new(
-             Config.cassandra_node!(),
-             cqex_options
-           ) do
-      Queries.get_jwt_public_key_pem(client)
-    else
-      {:error, :shutdown} ->
-        {:error, :realm_not_found}
-    end
+    Queries.get_jwt_public_key_pem(keyspace_name)
   end
 
   def update_jwt_public_key_pem(realm_name, jwt_public_key_pem) do

apps/astarte_realm_management/lib/astarte_realm_management/queries.ex

@@ -93,17 +93,13 @@ defmodule Astarte.RealmManagement.Queries do
     )
   """
 
-  @query_jwt_public_key_pem """
-    SELECT blobAsVarchar(value)
-    FROM kv_store
-    WHERE group='auth' AND key='jwt_public_key_pem';
-  """
-
   @query_insert_jwt_public_key_pem """
   INSERT INTO kv_store (group, key, value)
   VALUES ('auth', 'jwt_public_key_pem', varcharAsBlob(:pem));
   """
 
+  @keyspace_does_not_exist_regex ~r/Keyspace (.*) does not exist/
+
   defp create_one_object_columns_for_mappings(mappings) do
     for %Mapping{endpoint: endpoint, value_type: value_type} <- mappings do
       column_name = CQLUtils.endpoint_to_db_column_name(endpoint)
@@ -1091,17 +1087,66 @@ defmodule Astarte.RealmManagement.Queries do
     end
   end
 
-  def get_jwt_public_key_pem(client) do
-    query =
-      DatabaseQuery.new()
-      |> DatabaseQuery.statement(@query_jwt_public_key_pem)
+  def get_jwt_public_key_pem(keyspace) do
+    case Xandra.Cluster.run(:xandra, &do_get_jwt_public_key_pem(keyspace, &1)) do
+      {:ok, pem} ->
+        {:ok, pem}
 
-    with {:ok, result} <- DatabaseQuery.call(client, query),
-         ["system.blobasvarchar(value)": pem] <- DatabaseResult.head(result) do
-      {:ok, pem}
-    else
-      _ ->
-        {:error, :public_key_not_found}
+      {:error, %Xandra.ConnectionError{} = err} ->
+        _ =
+          Logger.warning("Database connection error #{Exception.message(err)}.",
+            tag: "database_connection_error"
+          )
+
+        {:error, :database_connection_error}
+
+      {:error, %Xandra.Error{} = err} ->
+        handle_xandra_error(err)
+    end
+  end
+
+  defp do_get_jwt_public_key_pem(keyspace_name, conn) do
+    query = """
+    SELECT blobAsVarchar(value)
+    FROM #{keyspace_name}.kv_store
+    WHERE group='auth' AND key='jwt_public_key_pem';
+    """
+
+    with {:ok, prepared} <- Xandra.prepare(conn, query),
+         {:ok, page} <-
+           Xandra.execute(conn, prepared, %{},
+             uuid_format: :binary,
+             consistency: :quorum
+           ) do
+      case Enum.to_list(page) do
+        [%{"system.blobasvarchar(value)": pem}] ->
+          {:ok, pem}
+
+        [] ->
+          {:error, :public_key_not_found}
+      end
+    end
+  end
+
+  defp handle_xandra_error(error) do
+    %Xandra.Error{message: message} = error
+
+    case Regex.run(@keyspace_does_not_exist_regex, message) do
+      [_message, keyspace] ->
+        Logger.warning("Keyspace #{keyspace} does not exist.",
+          tag: "realm_not_found"
+        )
+
+        {:error, :realm_not_found}
+
+      nil ->
+        _ =
+          Logger.warning(
+            "Database error, cannot get realm public key: #{Exception.message(error)}.",
+            tag: "database_error"
+          )
+
+        {:error, :database_error}
     end
   end
 

apps/astarte_realm_management/test/astarte_realm_management/queries_test.exs

@@ -560,10 +560,7 @@ defmodule Astarte.RealmManagement.QueriesTest do
   end
 
   test "get JWT public key PEM" do
-    DatabaseTestHelper.connect_to_test_database()
-    client = connect_to_test_realm("autotestrealm")
-
-    assert Queries.get_jwt_public_key_pem(client) ==
+    assert Queries.get_jwt_public_key_pem("autotestrealm") ==
              {:ok, DatabaseTestHelper.jwt_public_key_pem_fixture()}
   end
 
@@ -573,15 +570,15 @@ defmodule Astarte.RealmManagement.QueriesTest do
 
     new_pem = "not_exactly_a_PEM_but_will_do"
     assert Queries.update_jwt_public_key_pem(client, new_pem) == :ok
-    assert Queries.get_jwt_public_key_pem(client) == {:ok, new_pem}
+    assert Queries.get_jwt_public_key_pem("autotestrealm") == {:ok, new_pem}
 
     # Put the PEM fixture back
     assert Queries.update_jwt_public_key_pem(
              client,
              DatabaseTestHelper.jwt_public_key_pem_fixture()
            ) == :ok
 
-    assert Queries.get_jwt_public_key_pem(client) ==
+    assert Queries.get_jwt_public_key_pem("autotestrealm") ==
              {:ok, DatabaseTestHelper.jwt_public_key_pem_fixture()}
   end