Add basic device console

lib/nerves_hub_cli/api.ex

@@ -49,6 +49,138 @@ defmodule NervesHubCLI.API do
     |> resp()
   end
 
+  @doc """
+  Make a streaming HTTP request. Returns {:ok, pid} where pid will send chunks to the caller.
+
+  The caller will receive messages in the form:
+    {:chunk, data} - A chunk of response data
+    {:done, status} - The response is complete
+    {:error, reason} - An error occurred
+  """
+  def stream_request(verb, path, params, auth) do
+    caller = self()
+    url = URI.parse(endpoint() <> "/" <> URI.encode(path))
+
+    pid =
+      spawn_link(fn ->
+        do_stream_request(caller, verb, url, params, auth)
+      end)
+
+    {:ok, pid}
+  end
+
+  defp do_stream_request(caller, verb, url, params, auth) do
+    scheme = if url.scheme == "https", do: :https, else: :http
+    port = url.port || if scheme == :https, do: 443, else: 80
+
+    connect_opts =
+      if scheme == :https do
+        [
+          transport_opts: [
+            verify: :verify_peer,
+            cacerts: ca_certs(),
+            server_name_indication: to_charlist(url.host),
+            customize_hostname_check: [match_fun: :public_key.pkix_verify_hostname_match_fun(:https)]
+          ]
+        ]
+      else
+        []
+      end
+
+    case Mint.HTTP.connect(scheme, url.host, port, connect_opts) do
+      {:ok, conn} ->
+        body = Jason.encode!(params)
+        req_headers = stream_headers(auth, byte_size(body))
+        path_with_query = url.path || "/"
+
+        case Mint.HTTP.request(conn, String.upcase(to_string(verb)), path_with_query, req_headers, body) do
+          {:ok, conn, request_ref} ->
+            stream_response_loop(caller, conn, request_ref)
+
+          {:error, _conn, reason} ->
+            send(caller, {:error, reason})
+        end
+
+      {:error, reason} ->
+        send(caller, {:error, reason})
+    end
+  end
+
+  defp stream_headers(%{token: "nh" <> _ = token}, content_length) do
+    [
+      {"authorization", "token #{token}"},
+      {"content-type", "application/json"},
+      {"accept", "text/plain"},
+      {"content-length", to_string(content_length)}
+    ]
+  end
+
+  defp stream_headers(_, content_length) do
+    [
+      {"content-type", "application/json"},
+      {"accept", "text/plain"},
+      {"content-length", to_string(content_length)}
+    ]
+  end
+
+  defp stream_response_loop(caller, conn, request_ref) do
+    receive do
+      message ->
+        case Mint.HTTP.stream(conn, message) do
+          :unknown ->
+            stream_response_loop(caller, conn, request_ref)
+
+          {:ok, conn, responses} ->
+            case process_stream_responses(caller, responses, request_ref) do
+              :continue ->
+                stream_response_loop(caller, conn, request_ref)
+
+              :done ->
+                Mint.HTTP.close(conn)
+            end
+
+          {:error, _conn, reason, _responses} ->
+            send(caller, {:error, reason})
+        end
+    after
+      120_000 ->
+        send(caller, {:error, :timeout})
+    end
+  end
+
+  defp process_stream_responses(caller, responses, request_ref) do
+    Enum.reduce_while(responses, :continue, fn
+      {:status, ^request_ref, status}, _acc ->
+        if status >= 200 and status < 300 do
+          {:cont, :continue}
+        else
+          send(caller, {:error, {:http_status, status}})
+          {:halt, :done}
+        end
+
+      {:headers, ^request_ref, _headers}, acc ->
+        {:cont, acc}
+
+      {:data, ^request_ref, data}, acc ->
+        if byte_size(data) > 0 do
+          send(caller, {:chunk, data})
+        end
+
+        {:cont, acc}
+
+      {:done, ^request_ref}, _acc ->
+        send(caller, :done)
+        {:halt, :done}
+
+      {:error, ^request_ref, reason}, _acc ->
+        send(caller, {:error, reason})
+        {:halt, :done}
+
+      _other, acc ->
+        {:cont, acc}
+    end)
+  end
+
   def file_request(verb, path, file, params, auth) do
     content_length = :filelib.file_size(file)
     {:ok, pid} = Agent.start_link(fn -> 0 end)

lib/nerves_hub_cli/api/device.ex

@@ -96,6 +96,22 @@ defmodule NervesHubCLI.API.Device do
     API.request(:post, code_path, params, auth)
   end
 
+  @doc """
+  Send Elixir code to execute on a device's console with streaming response.
+
+  Returns a stream of console output chunks. The connection stays open until
+  the caller stops reading or the server closes.
+
+  Verb: POST
+  Path: /orgs/:org_name/products/:product_name/devices/:device_identifier/code
+  """
+  @spec console(String.t(), String.t(), String.t(), String.t(), NervesHubCLI.API.Auth.t()) ::
+          {:ok, pid()} | {:error, any()}
+  def console(org_name, product_name, device_identifier, body, %Auth{} = auth) do
+    code_path = Path.join(path(org_name, product_name, device_identifier), "code")
+    API.stream_request(:post, code_path, %{body: body, stream: true}, auth)
+  end
+
   @deprecated "use NervesHubCLI.API.DeviceCertificate.list/4 instead"
   def cert_list(org_name, product_name, device_identifier, %Auth{} = auth) do
     DeviceCertificate.list(org_name, product_name, device_identifier, auth)

lib/nerves_hub_cli/cli/device.ex

@@ -179,6 +179,16 @@ defmodule NervesHubCLI.CLI.Device do
       # From stdin
       echo "NervesHubLink.status()" | nh device code my-device-123
 
+  ## console
+
+  Open an interactive IEx console to a device. Send Elixir code and see the
+  output streamed back in real-time.
+
+      nh device console DEVICE_IDENTIFIER
+
+  Type Elixir expressions and press Enter to send them to the device.
+  Press Ctrl+C to exit.
+
   ## script list
 
   List available support scripts for a product.
@@ -189,6 +199,9 @@ defmodule NervesHubCLI.CLI.Device do
 
     * `--product` - (Optional) The product name.
 
+      This defaults to the NERVES_HUB_PRODUCT environment variable (if set) or
+      the global configuration via `nerves_hub config set product "product_name"`
+
   ## script send
 
   Send a support script to a device for execution.
@@ -276,12 +289,16 @@ defmodule NervesHubCLI.CLI.Device do
       ["code", identifier] ->
         code(org, product, identifier, opts)
 
+      ["console", identifier] ->
+        console(org, product, identifier)
+
       ["script", "list"] ->
         script_list(org, product)
 
       ["script", "send", identifier, name_or_id] ->
         script_send(org, product, identifier, name_or_id, opts)
 
+
       _ ->
         render_help()
     end
@@ -302,6 +319,7 @@ defmodule NervesHubCLI.CLI.Device do
       nh device cert create DEVICE_IDENTIFIER
       nh device cert import DEVICE_IDENTIFIER CERT_PATH
       nh device code DEVICE_IDENTIFIER
+      nh device console DEVICE_IDENTIFIER
       nh device script list
       nh device script send DEVICE_IDENTIFIER SCRIPT_NAME
 
@@ -627,6 +645,90 @@ defmodule NervesHubCLI.CLI.Device do
     Shell.info("")
   end
 
+  @spec console(String.t(), String.t(), String.t()) :: :ok
+  def console(org, product, identifier) do
+    auth = Shell.request_auth()
+
+    Shell.info("Connecting to device #{identifier}...")
+    Shell.info("Type Elixir expressions and press Enter to send. Press Ctrl+C to exit.\n")
+
+    console_loop(org, product, identifier, auth, nil)
+  end
+
+  defp console_loop(org, product, identifier, auth, stream_pid) do
+    # Kill any previous stream process and drain its output
+    if stream_pid && Process.alive?(stream_pid) do
+      Process.unlink(stream_pid)
+      Process.exit(stream_pid, :kill)
+      drain_console_output()
+    end
+
+    # Read user input
+    case IO.gets("iex> ") do
+      :eof ->
+        Shell.info("\nExiting console.")
+        :ok
+
+      {:error, reason} ->
+        Shell.render_error({:error, reason})
+
+      line when is_binary(line) ->
+        code = String.trim(line)
+
+        if code != "" do
+          # Start streaming request
+          case NervesHubCLI.API.Device.console(org, product, identifier, code, auth) do
+            {:ok, pid} ->
+              # Print output as it arrives
+              receive_console_output()
+              console_loop(org, product, identifier, auth, pid)
+
+            {:error, reason} ->
+              Shell.render_error({:error, reason})
+              console_loop(org, product, identifier, auth, nil)
+          end
+        else
+          console_loop(org, product, identifier, auth, stream_pid)
+        end
+    end
+  end
+
+  defp receive_console_output do
+    receive do
+      {:chunk, data} ->
+        IO.write(data)
+        receive_console_output()
+
+      :done ->
+        :ok
+
+      {:error, reason} ->
+        Shell.error("Stream error: #{inspect(reason)}")
+    after
+      # Wait a bit for initial response, then return to prompt
+      # The streaming will continue in background
+      5000 ->
+        :ok
+    end
+  end
+
+  defp drain_console_output do
+    receive do
+      {:chunk, data} ->
+        IO.write(data)
+        drain_console_output()
+
+      :done ->
+        :ok
+
+      {:error, _reason} ->
+        :ok
+    after
+      100 ->
+        :ok
+    end
+  end
+
   defp render_certs(identifier, certs) when is_list(certs) do
     Shell.info("\nDevice: #{identifier}")
     Shell.info("Certificates:")