Flake: HM module

lib/agent.ex

@@ -8,10 +8,19 @@ defmodule Frajtano.Agent do
 
   @impl true
   def init(_) do
-    {:ok,
-     %{
-       keys: %{}
-     }}
+    {
+      :ok,
+      %{
+        keys: %{}
+      },
+      {:continue, :init_peers}
+    }
+  end
+
+  @impl true
+  def handle_continue(:init_peers, state) do
+    for peer <- Application.fetch_env!(:frajtano, :initial_peers), do: {:ok, _} = Peer.start(peer)
+    {:noreply, state}
   end
 
   @impl true
@@ -26,10 +35,9 @@ defmodule Frajtano.Agent do
         with {:ok, idents} <- Peer.identities(peer),
              do: {:ok, {idents, peer}}
       end)
-      
 
     # Double :ok-wrapping because of Task.async_stream
-    idents = (for {:ok, {:ok, {idents, peer}}} <- idents, do: {idents, peer})
+    idents = for {:ok, {:ok, {idents, peer}}} <- idents, do: {idents, peer}
 
     {
       :reply,
@@ -61,7 +69,8 @@ defmodule Frajtano.Agent do
   end
 
   def sign(agent \\ __MODULE__, request) do
-    GenServer.call(agent, {:sign, request})
+    # Signing can take some time, as a password may need to be entered or similar
+    GenServer.call(agent, {:sign, request}, :infinity)
   end
 
   def add_peer(agent \\ __MODULE__, path) do

lib/frajtano.ex

@@ -17,10 +17,10 @@ defmodule Frajtano.Supervisor do
   @impl true
   def init(:ok) do
   	children = [
+  	  {DynamicSupervisor, name: Frajtano.Peer},
   	  Frajtano.Agent,
-  	  {Frajtano.Listener, [Application.fetch_env!(:frajtano, :listen_path)]},
   	  {Task.Supervisor, name: Frajtano.ClientSupervisor},
-  	  {DynamicSupervisor, name: Frajtano.Peer}
+  	  {Frajtano.Listener, [Application.fetch_env!(:frajtano, :listen_path)]},
   	]
 
   	Supervisor.init(children, strategy: :one_for_one)

lib/peer.ex

@@ -17,18 +17,8 @@ defmodule Frajtano.Peer do
     {:ok, %{conn: conn, clients: :queue.new(), buffer: <<>>}}
   end
 
-  @impl true
-  def handle_call(packet, from, %{conn: conn, clients: clients} = state) do
-    case :gen_tcp.send(conn, Proto.encode(packet)) do
-      :ok ->
-        {:noreply, %{state | clients: :queue.in(from, clients)}}
-
-      {:error, :closed} ->
-        {:noreply, state, {:continue, :closed}}
-
-      {:error, e} ->
-        raise(e)
-    end
+  def reply({client, ref}, msg) do
+    send(client, {ref, msg})
   end
 
   defp handle_messages(%{clients: clients, buffer: buffer} = state) do
@@ -38,42 +28,85 @@ defmodule Frajtano.Peer do
 
       {msg, buffer} ->
         {{:value, client}, clients} = :queue.out(clients)
-        GenServer.reply(client, {:ok, msg})
+        reply(client, {:ok, msg})
         handle_messages(%{state | clients: clients, buffer: buffer})
     end
   end
 
+  @impl true
+  def handle_info({:send, packet, from}, %{conn: conn, clients: clients} = state) do
+    case :gen_tcp.send(conn, Proto.encode(packet)) do
+      :ok ->
+        {:noreply, %{state | clients: :queue.in(from, clients)}}
+
+      {:error, e} ->
+        {:noreply, state, {:continue, {:error, e}}}
+    end
+  end
+
   @impl true
   def handle_info({:tcp, conn, msg}, %{conn: conn, buffer: buffer} = state) do
     :inet.setopts(conn, active: :once)
     buffer = buffer <> msg
 
-    handle_messages(%{ state | buffer: buffer })
+    handle_messages(%{state | buffer: buffer})
   end
 
   @impl true
   def handle_info({:tcp_closed, _}, state) do
-    {:noreply, state, {:continue, :closed}}
+    {:noreply, state, {:continue, {:error, :closed}}}
   end
 
   @impl true
-  def handle_continue(:closed, %{clients: clients}) do
+  def handle_info(:timeout, state) do
+    {:noreply, state, {:continue, {:error, :closed}}}
+  end
+
+  @impl true
+  def handle_continue({:error, e}, %{clients: clients}) do
     clients
     |> :queue.to_list()
-    |> Enum.each(&GenServer.reply(&1, {:error, :closed}))
+    |> Enum.each(&reply(&1, {:error, e}))
 
-    {:stop, :closed, %{}}
+    {:stop, {:error, e}, %{}}
   end
 
   def identities(peer) do
-    with {:ok, {:agent_identities_answer, identities}} <-
-           GenServer.call(peer, {:agentc_request_identities, nil}),
-         do: {:ok, identities}
+    ref = make_ref()
+    send(peer, {:send, {:agentc_request_identities, nil}, {self(), ref}})
+    # Needs to be less than the timeout in Frajtano.Agent.identities on the Task.async_stream call
+    # That's 5000 by default
+    timer = Process.send_after(peer, :timeout, 4500)
+
+    receive do
+      {^ref, msg} ->
+        Process.cancel_timer(timer)
+        msg
+    end
+    |> case do
+      {:ok, {:agent_identities_answer, identities}} -> {:ok, identities}
+      {:ok, {:agent_failure, nil}} -> {:error, :agent_failure}
+      {:ok, msg} -> raise("Unexpected message #{inspect(msg)}")
+      {:error, e} -> {:error, e}
+    end
   end
 
   def sign(peer, request) do
-    with {:ok, {:agent_sign_response, signature}} <-
-           GenServer.call(peer, {:agentc_sign_request, request}),
-         do: {:ok, signature}
+    # Signing may take some time, as a password may need to be entered or similar
+    # There is therefore no timeout
+    # If something requests identities afterwards, it will timeout, which also kills this signature request
+    # The SSH agent protocol strict ordering leaves fun problems with timeouts, as it turns out
+    ref = make_ref()
+    send(peer, {:send, {:agentc_sign_request, request}, {self(), ref}})
+
+    receive do
+      {^ref, msg} -> msg
+    end
+    |> case do
+      {:ok, {:agent_sign_response, signature}} -> {:ok, signature}
+      {:ok, {:agent_failure, nil}} -> {:error, :agent_failure}
+      {:ok, msg} -> raise("Unexpected message #{inspect(msg)}")
+      {:error, e} -> {:error, e}
+    end
   end
 end