Merge branch 'main' into the-credo-experiment

config/runtime.exs

@@ -348,6 +348,10 @@ if config_env() == :prod do
   end
 end
 
+# Set a default max firmware upload size of 200MB for all environments
+config :nerves_hub, NervesHub.Firmwares.Upload,
+  max_size: System.get_env("FIRMWARE_UPLOAD_MAX_SIZE", "200000000") |> String.to_integer()
+
 ##
 # SMTP settings.
 #

lib/nerves_hub/deployments/orchestrator.ex

@@ -15,6 +15,8 @@ defmodule NervesHub.Deployments.Orchestrator do
 
   alias NervesHub.Devices
   alias NervesHub.Devices.Device
+  alias NervesHub.Deployments
+  alias NervesHub.Deployments.Deployment
   alias NervesHub.Repo
   alias Phoenix.PubSub
   alias Phoenix.Socket.Broadcast
@@ -24,7 +26,7 @@ defmodule NervesHub.Deployments.Orchestrator do
   end
 
   def name(deployment_id) when is_integer(deployment_id) do
-    {:via, Registry, {NervesHub.Deployments, deployment_id}}
+    {:via, Registry, {Deployments, deployment_id}}
   end
 
   def name(deployment), do: name(deployment.id)
@@ -49,6 +51,11 @@ defmodule NervesHub.Deployments.Orchestrator do
   As devices update and reconnect, the new orchestrator is told that the update
   was successful, and the process is repeated.
   """
+  @decorate with_span("Deployments.Orchestrator.trigger_update#noop")
+  def trigger_update(%Deployment{is_active: false}) do
+    :ok
+  end
+
   @decorate with_span("Deployments.Orchestrator.trigger_update")
   def trigger_update(deployment) do
     :telemetry.execute([:nerves_hub, :deployment, :trigger_update], %{count: 1})
@@ -67,7 +74,7 @@ defmodule NervesHub.Deployments.Orchestrator do
     }
 
     devices =
-      Registry.select(NervesHub.Devices.Registry, [
+      Registry.select(Devices.Registry, [
         {{:_, :_, :"$1"}, match_conditions, [match_return]}
       ])
 

lib/nerves_hub/devices.ex

@@ -723,6 +723,9 @@ defmodule NervesHub.Devices do
           deployment_id: deployment.id
         }
 
+      {:error, :deployment_not_active, _device} ->
+        %UpdatePayload{update_available: false}
+
       {:error, :up_to_date, _device} ->
         %UpdatePayload{update_available: false}
 
@@ -818,6 +821,9 @@ defmodule NervesHub.Devices do
 
   def verify_update_eligibility(device, deployment, now \\ DateTime.utc_now()) do
     cond do
+      not deployment.is_active ->
+        {:error, :deployment_not_active, device}
+
       device_matches_deployment?(device, deployment) ->
         {:error, :up_to_date, device}
 

lib/nerves_hub/devices/connections.ex

@@ -128,4 +128,20 @@ defmodule NervesHub.Devices.Connections do
     |> distinct(:device_id)
     |> order_by([:device_id, desc: :last_seen_at])
   end
+
+  def clean_stale_connections() do
+    interval = Application.get_env(:nerves_hub, :device_last_seen_update_interval_minutes)
+    a_minute_ago = DateTime.shift(DateTime.utc_now(), minute: -(interval + 1))
+
+    DeviceConnection
+    |> where(status: :connected)
+    |> where([d], d.last_seen_at < ^a_minute_ago)
+    |> Repo.update_all(
+      set: [
+        status: :disconnected,
+        disconnected_at: DateTime.utc_now(),
+        disconnected_reason: "Stale connection"
+      ]
+    )
+  end
 end

lib/nerves_hub/firmwares.ex

@@ -147,7 +147,7 @@ defmodule NervesHub.Firmwares do
             Repo.rollback(error)
         end
       end,
-      timeout: 30_000
+      timeout: 60_000
     )
   end
 

lib/nerves_hub/firmwares/upload/s3.ex

@@ -12,7 +12,7 @@ defmodule NervesHub.Firmwares.Upload.S3 do
   def upload_file(source_path, %{"s3_key" => s3_key}) do
     source_path
     |> S3.Upload.stream_file()
-    |> S3.upload(bucket(), s3_key)
+    |> S3.upload(bucket(), s3_key, timeout: 60_000)
     |> ExAws.request()
     |> case do
       {:ok, _} -> :ok

lib/nerves_hub/workers/clean_device_connection_states.ex

@@ -3,9 +3,13 @@ defmodule NervesHub.Workers.CleanDeviceConnectionStates do
     max_attempts: 5,
     queue: :device
 
+  alias NervesHub.Devices
+  alias NervesHub.Devices.Connections
+
   @impl Oban.Worker
   def perform(_) do
-    NervesHub.Devices.clean_connection_states()
+    Devices.clean_connection_states()
+    Connections.clean_stale_connections()
 
     :ok
   end

lib/nerves_hub_web/dymanic_config_multipart.ex

@@ -0,0 +1,36 @@
+defmodule NervesHubWeb.DymanicConfigMultipart do
+  @moduledoc """
+  A wrapper around `Plug.Parsers.MULTIPART` which allows for the `:length` opt (max file size)
+  to be set during runtime.
+
+  This also restricts large file uploads to the firmware upload api route.
+
+  This can later be expanded to allow for different file size limits based on the organization.
+
+  Thank you to https://hexdocs.pm/plug/Plug.Parsers.MULTIPART.html#module-dynamic-configuration
+  for the inspiration.
+  """
+
+  @multipart Plug.Parsers.MULTIPART
+
+  def init(opts) do
+    opts
+  end
+
+  def parse(conn, "multipart", subtype, headers, opts) do
+    opts = @multipart.init([length: max_file_size(conn)] ++ opts)
+    @multipart.parse(conn, "multipart", subtype, headers, opts)
+  end
+
+  def parse(conn, _type, _subtype, _headers, _opts) do
+    {:next, conn}
+  end
+
+  defp max_file_size(conn) do
+    if String.match?(conn.request_path, ~r/^\/api\/orgs\/\w+\/products\/\w+\/firmwares$/) do
+      Application.get_env(:nerves_hub, NervesHub.Firmwares.Upload, [])[:max_size]
+    else
+      1_000_000
+    end
+  end
+end

lib/nerves_hub_web/endpoint.ex

@@ -71,11 +71,9 @@ defmodule NervesHubWeb.Endpoint do
 
   plug(
     Plug.Parsers,
-    parsers: [:urlencoded, :multipart, :json],
+    parsers: [:urlencoded, NervesHubWeb.DymanicConfigMultipart, :json],
     pass: ["*/*"],
-    # 1GB
-    length: 1_073_741_824,
-    json_decoder: Jason
+    json_decoder: Phoenix.json_library()
   )
 
   plug(Sentry.PlugContext)

lib/nerves_hub_web/live/dashboard/index.ex

@@ -93,7 +93,7 @@ defmodule NervesHubWeb.Live.Dashboard.Index do
         Devices.get_minimal_device_location_by_org_id_and_product_id(org.id, product.id)
 
       latest_firmwares =
-        Deployments.get_deployments_by_product(product.id)
+        Deployments.get_deployments_by_product(product)
         |> Enum.reduce(%{}, fn deployment, acc ->
           Map.put(acc, deployment.firmware.uuid, deployment.firmware.platform)
         end)

lib/nerves_hub_web/live/firmware.ex

@@ -47,7 +47,7 @@ defmodule NervesHubWeb.Live.Firmware do
       accept: ~w(.fw),
       max_entries: 1,
       auto_upload: true,
-      max_file_size: 200_000_000,
+      max_file_size: max_file_size(),
       progress: &handle_progress/3
     )
     |> render_with(&upload_firmware_template/1)
@@ -194,4 +194,8 @@ defmodule NervesHubWeb.Live.Firmware do
     key = Enum.find(org_keys, &(&1.id == org_key_id))
     "#{key.name}"
   end
+
+  defp max_file_size() do
+    Application.get_env(:nerves_hub, NervesHub.Firmwares.Upload, [])[:max_size]
+  end
 end

mix.lock

@@ -1,7 +1,7 @@
 %{
   "acceptor_pool": {:hex, :acceptor_pool, "1.0.0", "43c20d2acae35f0c2bcd64f9d2bde267e459f0f3fd23dab26485bf518c281b21", [:rebar3], [], "hexpm", "0cbcd83fdc8b9ad2eee2067ef8b91a14858a5883cb7cd800e6fcd5803e158788"},
   "assert_eventually": {:hex, :assert_eventually, "1.0.0", "f1539f28ba3ffa99a712433c77723c7103986932aa341d05eee94c333a920d15", [:mix], [{:ex_doc, ">= 0.0.0", [hex: :ex_doc, repo: "hexpm", optional: true]}], "hexpm", "c658ac4103c8bd82d0cf72a2fdb77477ba3fbc6b15228c5c801003d239625c69"},
-  "bandit": {:hex, :bandit, "1.6.2", "a5fa4cfbae9baaf196269a88533e18eef9e7c53bea07b03f6bc2c6d5bf87b1ce", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "4563b81ec94f25448ac02a8853453198cf7a63abac6202dbd4bda2c7f1a71eed"},
+  "bandit": {:hex, :bandit, "1.6.3", "36591efd4bcf0e0508c16aee42b574b6c374077f7b96575ff46c519c827db144", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "158a9802ec02ac297689948da8ce529a915528be11cb8fe0f27d1346864f50c0"},
   "base62": {:hex, :base62, "1.2.2", "85c6627eb609317b70f555294045895ffaaeb1758666ab9ef9ca38865b11e629", [:mix], [{:custom_base, "~> 0.2.1", [hex: :custom_base, repo: "hexpm", optional: false]}], "hexpm", "d41336bda8eaa5be197f1e4592400513ee60518e5b9f4dcf38f4b4dae6f377bb"},
   "bcrypt_elixir": {:hex, :bcrypt_elixir, "3.2.0", "feab711974beba4cb348147170346fe097eea2e840db4e012a145e180ed4ab75", [:make, :mix], [{:comeonin, "~> 5.3", [hex: :comeonin, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "563e92a6c77d667b19c5f4ba17ab6d440a085696bdf4c68b9b0f5b30bc5422b8"},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
@@ -103,7 +103,7 @@
   "telemetry_metrics_statsd": {:hex, :telemetry_metrics_statsd, "0.7.1", "3502235bb5b35ce50d608bf0f34369ef76eb92a4dbc8708c7e8780ca0da2d53e", [:mix], [{:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "06338d9dc3b4a202f11a6e706fd3feba4c46100d0aca23688dea0b8f801c361f"},
   "telemetry_poller": {:hex, :telemetry_poller, "1.1.0", "58fa7c216257291caaf8d05678c8d01bd45f4bdbc1286838a28c4bb62ef32999", [:rebar3], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9eb9d9cbfd81cbd7cdd24682f8711b6e2b691289a0de6826e58452f28c103c8f"},
   "telemetry_registry": {:hex, :telemetry_registry, "0.3.2", "701576890320be6428189bff963e865e8f23e0ff3615eade8f78662be0fc003c", [:mix, :rebar3], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e7ed191eb1d115a3034af8e1e35e4e63d5348851d556646d46ca3d1b4e16bab9"},
-  "thousand_island": {:hex, :thousand_island, "1.3.8", "18399faae8e09b38254af188d8ba88fa2921806cc9d622e157f74e1ffaefada2", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6b2aaff981345de0f1532654998f17b9ba46489b63d1c1560e3c6589d2a81d0c"},
+  "thousand_island": {:hex, :thousand_island, "1.3.9", "095db3e2650819443e33237891271943fad3b7f9ba341073947581362582ab5a", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "25ab4c07badadf7f87adb4ab414e0ed374e5f19e72503aa85132caa25776e54f"},
   "timex": {:hex, :timex, "3.7.11", "bb95cb4eb1d06e27346325de506bcc6c30f9c6dea40d1ebe390b262fad1862d1", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:gettext, "~> 0.20", [hex: :gettext, repo: "hexpm", optional: false]}, {:tzdata, "~> 1.1", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm", "8b9024f7efbabaf9bd7aa04f65cf8dcd7c9818ca5737677c7b76acbc6a94d1aa"},
   "tls_certificate_check": {:hex, :tls_certificate_check, "1.24.0", "d00e2887551ff8cdae4d0340d90d9fcbc4943c7b5f49d32ed4bc23aff4db9a44", [:rebar3], [{:ssl_verify_fun, "~> 1.1", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm", "90b25a58ee433d91c17f036d4d354bf8859a089bfda60e68a86f8eecae45ef1b"},
   "tzdata": {:hex, :tzdata, "1.1.2", "45e5f1fcf8729525ec27c65e163be5b3d247ab1702581a94674e008413eef50b", [:mix], [{:hackney, "~> 1.17", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "cec7b286e608371602318c414f344941d5eb0375e14cfdab605cca2fe66cba8b"},

test/nerves_hub/devices_test.exs

@@ -22,7 +22,7 @@ defmodule NervesHub.DevicesTest do
     product = Fixtures.product_fixture(user, org)
     org_key = Fixtures.org_key_fixture(org, user)
     firmware = Fixtures.firmware_fixture(org_key, product)
-    deployment = Fixtures.deployment_fixture(org, firmware)
+    deployment = Fixtures.deployment_fixture(org, firmware, %{is_active: true})
     device = Fixtures.device_fixture(org, product, firmware)
     device2 = Fixtures.device_fixture(org, product, firmware)
     device3 = Fixtures.device_fixture(org, product, firmware)

test/nerves_hub_web/live/devices/show_test.exs

@@ -374,7 +374,7 @@ defmodule NervesHubWeb.Live.Devices.ShowTest do
       firmware = Fixtures.firmware_fixture(org_key, product, %{dir: tmp_dir})
 
       deployment
-      |> Ecto.Changeset.change(%{firmware_id: firmware.id})
+      |> Ecto.Changeset.change(%{firmware_id: firmware.id, is_active: true})
       |> Repo.update!()
 
       NervesHubWeb.Endpoint.subscribe("device:#{device.id}")
@@ -390,6 +390,36 @@ defmodule NervesHubWeb.Live.Devices.ShowTest do
 
       assert_receive %Phoenix.Socket.Broadcast{event: "deployments/update"}
     end
+
+    test "available update exists but deployment is not active", %{
+      conn: conn,
+      org: org,
+      product: product,
+      device: device,
+      deployment: deployment,
+      org_key: org_key,
+      tmp_dir: tmp_dir
+    } do
+      device =
+        device
+        |> Ecto.Changeset.change(%{deployment_id: deployment.id})
+        |> Repo.update!()
+
+      firmware = Fixtures.firmware_fixture(org_key, product, %{dir: tmp_dir})
+
+      deployment
+      |> Ecto.Changeset.change(%{firmware_id: firmware.id, is_active: false})
+      |> Repo.update!()
+
+      NervesHubWeb.Endpoint.subscribe("device:#{device.id}")
+
+      conn
+      |> visit("/org/#{org.name}/#{product.name}/devices/#{device.identifier}")
+      |> assert_has("h1", text: device.identifier)
+      |> refute_has("span", text: "Update available")
+
+      assert Repo.aggregate(NervesHub.Devices.InflightUpdate, :count) == 0
+    end
   end
 
   describe "support scripts" do

test/support/fixtures.ex

@@ -215,12 +215,16 @@ defmodule NervesHub.Fixtures do
   end
 
   def deployment_fixture(%Org{} = org, %Firmwares.Firmware{} = firmware, params \\ %{}) do
+    {is_active, params} = Map.pop(params, :is_active, false)
+
     {:ok, deployment} =
       %{org_id: org.id, firmware_id: firmware.id}
       |> Enum.into(params)
       |> Enum.into(@deployment_params)
       |> Deployments.create_deployment()
 
+    {:ok, deployment} = Deployments.update_deployment(deployment, %{is_active: is_active})
+
     deployment
   end