Merge pull request #2 from daveminer/build-handshake-msg

Build handshake msg

.github/workflows/ci.yml

@@ -0,0 +1,42 @@
+name: Elixir CI
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        elixir-version: [1.16, 1.17, 1.18]
+        otp-version: [24.0, 25.0, 26.0]
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Elixir
+        uses: erlef/setup-beam@v1
+        with:
+          elixir-version: ${{ matrix.elixir-version }}
+          otp-version: ${{ matrix.otp-version }}
+
+      - name: Cache dependencies
+        uses: actions/cache@v3
+        with:
+          path: deps
+          key: ${{ runner.os }}-mix-${{ hashFiles('**/mix.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-mix-
+
+      - name: Install dependencies
+        run: mix deps.get
+
+      - name: Run tests
+        run: mix test

lib/mix/tasks/dev.ex

@@ -0,0 +1,16 @@
+defmodule Mix.Tasks.Dev do
+  use Mix.Task
+
+  alias Rex.Handshake
+
+  def run(_) do
+    Application.ensure_all_started(:rex)
+
+    msg = Handshake.Proposal.version_message([10, 11, 12, 13, 14, 15, 16], :mainnet)
+
+    dbg(msg)
+    dbg(CBOR.decode(msg))
+
+    :ok
+  end
+end

lib/rex.ex

@@ -2,6 +2,11 @@ defmodule Rex do
   @moduledoc false
 
   def get_current_era do
+    # Rex.Client.query(:get_current_era) |> IO.inspect()
+    # Rex.Client.query(:get_current_era) |> IO.inspect()
+    # Rex.Client.query(:get_current_era) |> IO.inspect()
+    Rex.ClientStatem.query(:get_current_era) |> IO.inspect()
+    Rex.ClientStatem.query(:get_current_era) |> IO.inspect()
     Rex.ClientStatem.query(:get_current_era) |> IO.inspect()
   end
 end

lib/rex/application.ex

@@ -13,16 +13,18 @@ defmodule Rex.Application do
     Supervisor.start_link(children, opts)
   end
 
-  @default_node_socket_path "/tmp/cardano_node.socket"
-
   defp client_opts do
-    dbg(System.get_env("CARDANO_NODE_SOCKET_PATH"))
+    network = System.get_env("CARDANO_NETWORK", "mainnet") |> String.to_atom()
+    path = System.get_env("CARDANO_NODE_PATH", "/tmp/cardano-node.socket")
+    port = System.get_env("CARDANO_NODE_PORT", "0") |> String.to_integer()
+    # socket, tcp or tls
+    type = System.get_env("CARDANO_NODE_TYPE", "socket")
 
     [
-      node_port: System.get_env("CARDANO_NODE_PORT", "9443") |> String.to_integer(),
-      node_url: System.get_env("CARDANO_NODE_URL"),
-      socket_path: System.get_env("CARDANO_NODE_SOCKET_PATH"),
-      network: :mainnet
+      network: network,
+      path: path,
+      port: port,
+      type: type
     ]
   end
 end

lib/rex/client_statem.ex

@@ -4,15 +4,16 @@ defmodule Rex.ClientStatem do
   """
   @behaviour :gen_statem
 
-  alias Rex.HandshakeResponse
+  alias Rex.Handshake
   alias Rex.LocalStateQueryResponse
   alias Rex.Messages
 
   require Logger
 
   @basic_tcp_opts [:binary, active: false, send_timeout: 4_000]
+  @active_n2c_versions [9, 10, 11, 12, 13, 14, 15, 16]
 
-  defstruct [:node_port, :node_url, :path, :socket, :network, queue: :queue.new()]
+  defstruct [:client, :path, :port, :socket, :network, queue: :queue.new()]
 
   ##############
   # Public API #
@@ -22,12 +23,12 @@ defmodule Rex.ClientStatem do
     :gen_statem.call(pid, {:request, query_name})
   end
 
-  def start_link(opts) do
+  def start_link(network: network, path: path, port: port, type: type) do
     data = %__MODULE__{
-      node_port: Keyword.fetch!(opts, :node_port),
-      node_url: Keyword.fetch!(opts, :node_url),
-      path: Keyword.fetch!(opts, :socket_path),
-      network: Keyword.get(opts, :network, :mainnet),
+      client: tcp_lib(type),
+      path: maybe_local_path(path, type),
+      port: maybe_local_port(port, type),
+      network: network,
       socket: nil
     }
 
@@ -57,12 +58,15 @@ defmodule Rex.ClientStatem do
     {:ok, :disconnected, data, actions}
   end
 
-  def disconnected(:internal, :connect, %__MODULE__{node_url: node_url, path: path} = data) do
-    # Connect to local unix socket on `path`
-    case tcp_lib(data.path).connect(
-           node_path(%{socket_path: path, node_url: node_url}),
-           node_port(data.path),
-           tcp_opts(data.path, node_url)
+  def disconnected(
+        :internal,
+        :connect,
+        %__MODULE__{client: client, path: path, port: port} = data
+      ) do
+    case client.connect(
+           maybe_parse_path(path),
+           port,
+           tcp_opts(client, path)
          ) do
       {:ok, socket} ->
         data = %__MODULE__{data | socket: socket}
@@ -80,13 +84,21 @@ defmodule Rex.ClientStatem do
     {:keep_state, data, actions}
   end
 
-  def connected(:internal, :establish, %__MODULE{socket: socket, network: network} = data) do
-    :ok = tcp_lib(data.path).send(socket, Messages.handshake(network))
+  def connected(
+        :internal,
+        :establish,
+        %__MODULE__{client: client, socket: socket, network: network} = data
+      ) do
+    :ok =
+      client.send(
+        socket,
+        Handshake.Proposal.version_message(@active_n2c_versions, network)
+      )
 
-    case tcp_lib(data.path).recv(socket, 0, 5_000) do
+    case client.recv(socket, 0, 5_000) do
       {:ok, full_response} ->
-        {:ok, handshake} = HandshakeResponse.parse_response(full_response)
-        IO.inspect(handshake)
+        {:ok, _handshake_response} = Handshake.Response.validate(full_response)
+
         actions = [{:next_event, :internal, :acquire_agency}]
         {:next_state, :established_no_agency, data, actions}
 
@@ -96,9 +108,13 @@ defmodule Rex.ClientStatem do
     end
   end
 
-  def established_no_agency(:internal, :acquire_agency, %__MODULE__{socket: socket} = data) do
-    :ok = tcp_lib(data.path).send(socket, Messages.msg_acquire())
-    {:ok, _acquire_response} = tcp_lib(data.path).recv(socket, 0, 5_000)
+  def established_no_agency(
+        :internal,
+        :acquire_agency,
+        %__MODULE__{client: client, socket: socket} = data
+      ) do
+    :ok = client.send(socket, Messages.msg_acquire())
+    {:ok, _acquire_response} = client.recv(socket, 0, 5_000)
     {:next_state, :established_has_agency, data}
   end
 
@@ -110,10 +126,10 @@ defmodule Rex.ClientStatem do
   def established_has_agency(
         {:call, from},
         {:request, :get_current_era},
-        %__MODULE__{socket: socket} = data
+        %__MODULE__{client: client, socket: socket} = data
       ) do
-    :ok = setopts_lib(data.path).setopts(socket, active: :once)
-    :ok = tcp_lib(data.path).send(socket, Messages.get_current_era())
+    :ok = setopts_lib(client).setopts(socket, active: :once)
+    :ok = client.send(socket, Messages.get_current_era())
     data = update_in(data.queue, &:queue.in(from, &1))
     {:keep_state, data}
   end
@@ -135,27 +151,29 @@ defmodule Rex.ClientStatem do
     {:next_state, :disconnected, data}
   end
 
-  defp node_path(%{socket_path: nil, node_url: nil}), do: raise("No node path or URL provided")
-  defp node_path(%{socket_path: nil, node_url: url}), do: ~c"#{url}"
-  defp node_path(%{socket_path: path, node_url: _}), do: {:local, ~c"#{path}"}
+  defp maybe_local_path(path, "socket"), do: {:local, path}
+  defp maybe_local_path(path, _), do: path
+
+  defp maybe_local_port(_port, "socket"), do: 0
+  defp maybe_local_port(port, _), do: port
 
-  defp node_port(nil), do: 9443
-  defp node_port(_), do: 0
+  defp maybe_parse_path(path) when is_binary(path), do: ~c[#{path}]
+  defp maybe_parse_path(path), do: path
 
-  defp tcp_lib(nil), do: :ssl
+  defp tcp_lib("ssl"), do: :ssl
   defp tcp_lib(_), do: :gen_tcp
 
-  defp tcp_opts(nil, url),
+  defp tcp_opts(:ssl, path),
     do:
       @basic_tcp_opts ++
         [
           verify: :verify_none,
-          server_name_indication: ~c"#{url}",
+          server_name_indication: ~c"#{path}",
           secure_renegotiate: true
         ]
 
   defp tcp_opts(_, _), do: @basic_tcp_opts
 
-  defp setopts_lib(nil), do: :ssl
+  defp setopts_lib(:ssl), do: :ssl
   defp setopts_lib(_), do: :inet
 end

lib/rex/handshake/proposal.ex

@@ -0,0 +1,64 @@
+defmodule Rex.Handshake.Proposal do
+  @moduledoc """
+  Builds handshake messages for node-to-client communication.
+  """
+
+  @type network_type :: :mainnet | :preprod | :preview | :sanchonet
+
+  @network_magic [
+    mainnet: 764_824_073,
+    preprod: 1,
+    preview: 2,
+    sanchonet: 4
+  ]
+
+  @version_numbers %{
+    9 => 32777,
+    10 => 32778,
+    11 => 32779,
+    12 => 32780,
+    13 => 32781,
+    14 => 32782,
+    15 => 32783,
+    16 => 32784,
+    17 => 32785
+  }
+
+  @doc """
+  Version numbers must be unique and appear in ascending order.
+  """
+  @spec version_message([integer()], network_type) :: binary()
+  def version_message(versions, network) do
+    payload =
+      [
+        # msgProposeVersions
+        0,
+        build_version_fragments(versions |> Enum.sort(), network)
+      ]
+      |> CBOR.encode()
+
+    header(payload) <> payload
+  end
+
+  defp build_version_fragments(versions, network),
+    do:
+      Enum.reduce(versions, %{}, fn version, acc ->
+        Map.merge(acc, version_fragment(version, network))
+      end)
+
+  defp version_fragment(version, network) when version >= 15,
+    do: %{@version_numbers[version] => [@network_magic[network], false]}
+
+  defp version_fragment(version, network),
+    do: %{@version_numbers[version] => @network_magic[network]}
+
+  # middle 16 bits are: 1 bit == 0 for initiator and 15 bits for the mini protocol ID (0)
+  defp header(payload),
+    do: <<header_timestamp()::32>> <> <<0, 0>> <> <<byte_size(payload)::unsigned-16>>
+
+  # Returns the lower 32 bits of the system's monotonic time in microseconds
+  defp header_timestamp(),
+    do:
+      System.monotonic_time(:microsecond)
+      |> Bitwise.band(0xFFFFFFFF)
+end

lib/rex/handshake/response.ex

@@ -0,0 +1,47 @@
+defmodule Rex.Handshake.Response do
+  defstruct [:type, :version_number, :network_magic, :query]
+
+  alias Rex.Util
+
+  def validate(response) do
+    %{payload: payload} = Util.plex(response)
+
+    case CBOR.decode(payload) do
+      # msgAcceptVersion
+      {:ok, [1, version, [magic, query]], ""} ->
+        if version in [32783, 32784] do
+          {:ok,
+           %__MODULE__{
+             network_magic: magic,
+             query: query,
+             type: :msg_accept_version,
+             version_number: version
+           }}
+        else
+          {:error, "Only versions 32783 and 32784 are supported."}
+        end
+
+      # msgRefuse
+      {:ok, [2, refuse_reason], ""} ->
+        case refuse_reason do
+          # TODO: return accepted versions; reduce to 32783 and 32784
+          [0, _version_number_binary] ->
+            {:refused, %__MODULE__{type: :version_mismatch}}
+
+          [1, _anyVersionNumber, _tstr] ->
+            {:refused, %__MODULE__{type: :handshake_decode_error}}
+
+          [2, _anyVersionNumber, _tstr] ->
+            {:refused, %__MODULE__{type: :refused}}
+        end
+
+      # TODO: parse version_table
+      # msgQueryReply
+      {:ok, [3, version_table], ""} ->
+        {:versions, version_table}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+end