Security, correctness and dead code fixes

.github/workflows/main.yml

@@ -61,3 +61,15 @@ jobs:
         run: mix compile --warnings-as-errors
       - name: Run Tests
         run: mix test
+
+  format:
+    runs-on: ubuntu-24.04
+    name: Format
+    steps:
+      - uses: actions/checkout@v4
+      - uses: erlef/setup-beam@v1
+        with:
+          otp-version: "28"
+          elixir-version: "1.20"
+      - name: Check formatting
+        run: mix format --check-formatted

CHANGELOG.md

@@ -4,6 +4,22 @@
 
 - **[BREAKING]** Updated the minimum required version of Elixir to `1.16`
 - Fix type warnings emitted by `use Storex.Store` on Elixir 1.18 and newer
+- **[BREAKING]** A `FunctionClauseError` raised inside a mutation that *did* match is no longer reported to the client as `No mutation matching ...`. Only the store's own `mutation/5` failing to match produces that error; anything else propagates with its original stacktrace, which stops the store process and the socket with it
+- `terminate/3` is no longer skipped when the store module happens not to be loaded yet. `Storex.Store.__terminate__/4` gated the call on `function_exported?/3`, which answers `false` for an unloaded module, so whether the callback ran depended on the code server rather than on the store
+- Housekeeping: dropped the unused `import Supervisor.Spec, warn: false` from `Storex.start/2` (deprecated since Elixir 1.5, and the `warn: false` was hiding it) and the `registry: Storex.Registry.ETS` key from `config/config.exs` (that module was removed in 0.4.0 and nothing read the key). The project is now formatted, and CI checks it
+- `Storex.Registry` reads (`get_store/2`, `get_store_pid/2`, `get_store_instances/1`, `session_stores/1`) now run in the calling process against the `:protected` ETS table instead of a `GenServer.call`. The registry process was a global serialisation point — every mutation performs at least one lookup — and the cost grew with the number of connections. Measured at 500 lookups per reader: 98.3ms against 16.2ms with 64 concurrent readers, 335.8ms against 56.2ms with 256. Writes and the `:DOWN` cleanup still go through the process
+- **[BREAKING]** `Storex.mutate/3` and `Storex.mutate/4` return `:ok`. They used to return whatever the comprehension in `Storex.PG.broadcast/1` produced, which was the internal broadcast envelope repeated once per node (`[broadcast: {:mutate, "Store", "reload", []}]`)
+- Removed the unreachable `{:error, _}` branch in `Storex.PG.broadcast/1`. `:pg.get_members/2` always returns a list; the error tuple was `:pg2`'s contract, and `:pg2` support went away in this release
+- The state a store reports on join is now read by asking the store process (`handle_call(:get_state, ...)` on the generated `Server`) instead of `:sys.get_state/1`, a debug function that was reaching into the process's internal state shape from the outside on every join. If the process is gone by the time the join reads it, the client now gets an error frame instead of the connection exiting
+- A client could kill its connection process with a well-formed `error` frame: `Storex.Message.cast/1` accepted the shape but `Storex.Socket.message_handle/2` had no clause for it. `error` frames only travel server to client, so the shape is no longer accepted and the frame is refused with `1007` like any other unknown type
+- `Storex.Handler.Plug` closed a malformed payload with a bare `1007`, passing the reason as the process exit reason instead of the close payload. It now sends `1007` with the reason, matching `Storex.Handler.Cowboy`
+- Removed dead code: `Storex.Registry.session_pid/1` (no callers, and its `:ets.match/2` pattern was a 4-tuple against 5-tuple records, so it never matched anything) and `Storex.Handler.Cowboy.websocket_init/3` (a cowboy 1.x callback, unreachable on cowboy 2.x)
+- **[SECURITY]** The SSR/HTTP path now runs the same store-name checks as the WebSocket path. `Storex.HTTP` carried its own copy of the resolution logic with only `Module.safe_concat/1`, missing `Code.ensure_compiled/1` and the `Storex.Store` behaviour check, so `GET /storex?store=Any.Module&params=%7B%7D` called `init/2` on any loaded module whose name resolved and serialised the result back to the caller. Both transports now share `Storex.Store.resolve/1`
+- **[BREAKING]** **[SECURITY]** `Storex.Handler.Cowboy` no longer accepts `:binary` frames. It decoded them with `:erlang.binary_to_term/1`, which creates atoms out of bytes the client controls — a single 25-byte frame is enough — and passed the resulting term straight to `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1` allowlist every other entry point goes through. Binary frames are now closed with `1003`
+- `Storex.Handler.Plug` had no clause for `:binary` frames at all, so one raised `FunctionClauseError` and closed the connection with `1011`. It now closes with `1003` like the cowboy handler
+- **[SECURITY]** A `mutation` frame is now resolved against the session the server assigned to the connection, not the `session` field carried by the frame. Any client could previously mutate — and read the resulting diff of — any other session's store by naming its session id. Mutating other sessions on purpose is what `Storex.mutate/3` and `Storex.mutate/4` are for
+- A `mutation` for a store the session has not joined now returns an error to the client instead of exiting the connection process
+- Store processes are now registered through a `Registry` keyed by `{session, store}` instead of being named `:"#{session}_#{store}"`. Session ids are unique per connection, so the old naming created one permanent atom per session-store pair and could exhaust the atom table on a long-running node
 - Removed the `:pg2` fallback, unreachable since OTP 24
 - Updated dependencies
 

config/config.exs

@@ -1,8 +1,6 @@
 import Config
 
-config :storex,
-  session_id_library: Nanoid,
-  registry: Storex.Registry.ETS
+config :storex, session_id_library: Nanoid
 
 if Mix.env() == :test do
   import_config "test.exs"

lib/storex.ex

@@ -3,12 +3,11 @@ defmodule Storex do
 
   @doc false
   def start(_type, _args) do
-    import Supervisor.Spec, warn: false
-
     children = [
       %{id: :pg, start: {:pg, :start_link, [Storex.PG]}},
       {Storex.PG, []},
       {Storex.Registry, []},
+      {Registry, keys: :unique, name: Storex.StoreRegistry},
       {Storex.Supervisor, []}
     ]
 

lib/storex/handler/cowboy.ex

@@ -9,10 +9,6 @@ defmodule Storex.Handler.Cowboy do
     {:cowboy_websocket, request, %{session: session, pid: request.pid}}
   end
 
-  def websocket_init(_type, req, _opts) do
-    {:ok, req, %{status: "inactive"}}
-  end
-
   def terminate(_reason, _req, %{session: session}) do
     Storex.Registry.session_stores(session)
     |> Enum.each(fn {store, _, session, _, _} ->
@@ -26,14 +22,13 @@ defmodule Storex.Handler.Cowboy do
     :ok
   end
 
-  def websocket_handle({:binary, frame}, state) do
-    try do
-      :erlang.binary_to_term(frame)
-      |> Socket.message_handle(state)
-      |> map_response()
-    rescue
-      ArgumentError -> {:reply, {:close, 1007, "Payload is malformed."}, state}
-    end
+  # Binary frames are not part of the protocol: the client only ever sends text.
+  # They used to be decoded with `:erlang.binary_to_term/1`, which creates atoms
+  # out of bytes the client controls and handed the resulting term straight to
+  # `Storex.Socket.message_handle/2`, bypassing the `Storex.Message.cast/1`
+  # allowlist that every other entry point goes through.
+  def websocket_handle({:binary, _frame}, state) do
+    {:reply, {:close, 1003, "Binary frames are not supported."}, state}
   end
 
   def websocket_handle({:text, frame}, state) do

lib/storex/handler/plug.ex

@@ -30,10 +30,16 @@ defmodule Storex.Handler.Plug do
       |> map_response()
     else
       {:error, _} ->
-        {:stop, "Payload is malformed.", 1007, state}
+        {:stop, :normal, {1007, "Payload is malformed."}, state}
     end
   end
 
+  # See the note in `Storex.Handler.Cowboy`. Without this clause a binary frame
+  # raises `FunctionClauseError` here and takes the connection down with a 1011.
+  def handle_in({_message, [opcode: :binary]}, state) do
+    {:stop, :normal, {1003, "Binary frames are not supported."}, state}
+  end
+
   def handle_info({:mutate, store, mutation, data}, %{session: session} = state) do
     %{
       type: "mutation",

lib/storex/http.ex

@@ -1,6 +1,6 @@
 defmodule Storex.HTTP do
   def init_store(store, params) do
-    with {:store, {:ok, store_module}} <- {:store, store |> get_module()},
+    with {:store, {:ok, store_module}} <- {:store, Storex.Store.resolve(store)},
          {:params, {:ok, params}} <- {:params, params |> get_params()},
          {:state, {:ok, result}} <- {:state, get_state(store_module, params)} do
       {:ok,
@@ -17,7 +17,7 @@ defmodule Storex.HTTP do
            type: "error",
            session: "SSR",
            store: store,
-           error: "Store '#{inspect(store)}' is not defined or can't be compiled."
+           error: "Store '#{store}' is not defined or can't be compiled."
          }}
 
       {:state, {:error, message}} ->
@@ -40,35 +40,20 @@ defmodule Storex.HTTP do
     end
   end
 
-  defp get_module(store) do
-    try do
-      module = Module.safe_concat([store])
-      {:ok, module}
-    rescue
-      ArgumentError -> {:error, :not_exists}
-    end
-  end
-
   defp get_params(params) do
     params
     |> Jason.decode()
   end
 
+  # The SSR path is init-only, so the key a store may return is dropped. Using
+  # the shared dispatcher keeps the accepted return values, and the error raised
+  # for anything else, identical to the websocket path.
   defp get_state(module, params) do
     module
-    |> apply(:init, ["SSR", params])
-    |> result()
-  end
-
-  defp result({:ok, state}) do
-    {:ok, state}
-  end
-
-  defp result({:ok, state, _}) do
-    {:ok, state}
-  end
-
-  defp result({:error, error_message}) do
-    {:error, error_message}
+    |> Storex.Store.__init__("SSR", params)
+    |> case do
+      {:ok, state, _key} -> {:ok, state}
+      {:error, reason} -> {:error, reason}
+    end
   end
 end

lib/storex/message.ex

@@ -37,17 +37,10 @@ defmodule Storex.Message do
      }}
   end
 
-  def cast(%{
-        "type" => "error",
-        "store" => store,
-        "data" => data,
-        "request" => request,
-        "session" => session
-      }) do
-    {:ok,
-     %__MODULE__{type: "error", store: store, data: data, request: request, session: session}}
-  end
-
+  # `error` frames only ever travel server to client, and are built as plain maps
+  # in `Storex.Socket`. Casting one here made it past the allowlist and then hit
+  # `Storex.Socket.message_handle/2`, which has no clause for it, so a client
+  # could kill its connection process with a well-formed frame.
   def cast(_) do
     {:error, "Unknown message type"}
   end

lib/storex/pg.ex

@@ -14,17 +14,15 @@ defmodule Storex.PG do
     {:ok, @name}
   end
 
+  # `:pg.get_members/2` always returns a list — the `{:error, _}` clause this used
+  # to carry was `:pg2`'s contract, and `:pg2` is gone. `Enum.each/2` rather than
+  # a comprehension so the return value is `:ok`: `send/2` returns the message,
+  # so `Storex.mutate/3` used to hand back the internal broadcast envelope once
+  # per node.
   def broadcast(payload) do
-    :pg.get_members(Storex.PG, @name)
-    |> case do
-      {:error, _} ->
-        :error
-
-      pids ->
-        for pid <- pids do
-          send(pid, {:broadcast, payload})
-        end
-    end
+    Storex.PG
+    |> :pg.get_members(@name)
+    |> Enum.each(&send(&1, {:broadcast, payload}))
   end
 
   @impl true

lib/storex/registry.ex

@@ -15,10 +15,6 @@ defmodule Storex.Registry do
     {:ok, %{}}
   end
 
-  def session_pid(session) do
-    GenServer.call(@registry, {:session_pid, session})
-  end
-
   def register_store(store, store_pid, session, session_pid, key) do
     GenServer.call(@registry, {:register_store, store, store_pid, session, session_pid, key})
   end
@@ -27,28 +23,35 @@ defmodule Storex.Registry do
     GenServer.call(@registry, {:unregister_store, store, session})
   end
 
+  # The table is `:protected`: only the owning process writes, but every process
+  # reads. So reads run in the caller. Routing them through this GenServer made
+  # it a global serialisation point for the whole library — every mutation does
+  # at least one lookup — and the cost scales with the number of connections.
+  # Measured, 500 lookups per reader: with 64 concurrent readers, 98.3ms through
+  # the GenServer against 16.2ms reading directly; with 256, 335.8ms against
+  # 56.2ms. Writes and the `:DOWN` cleanup stay in the process.
   def get_store(store, session) do
-    GenServer.call(@registry, {:get_store, store, session})
+    :ets.match_object(@registry, {store, :"$1", session, :_, :_})
+    |> case do
+      [] -> :undefined
+      [object | _tail] -> object
+    end
   end
 
   def get_store_pid(store, session) do
-    GenServer.call(@registry, {:get_store_pid, store, session})
+    :ets.match(@registry, {store, :"$1", session, :_, :_})
+    |> case do
+      [] -> :undefined
+      [[pid] | _tail] -> pid
+    end
   end
 
   def get_store_instances(query) do
-    GenServer.call(@registry, {:get_store_instances, query})
+    :ets.match_object(@registry, query)
   end
 
   def session_stores(session) do
-    GenServer.call(@registry, {:session_stores, session})
-  end
-
-  def handle_call({:session_pid, session}, _from, state) do
-    :ets.match(@registry, {:_, session, :_, :"$1"})
-    |> case do
-      [] -> {:reply, :undefined, state}
-      [[pid] | _tail] -> {:reply, pid, state}
-    end
+    :ets.match_object(@registry, {:_, :_, session, :_, :_})
   end
 
   def handle_call({:register_store, store, store_pid, session, session_pid, key}, _from, state) do
@@ -62,37 +65,9 @@ defmodule Storex.Registry do
     {:reply, result, state}
   end
 
-  def handle_call({:get_store, store, session}, _from, state) do
-    :ets.match_object(@registry, {store, :"$1", session, :_, :_})
-    |> case do
-      [] -> {:reply, :undefined, state}
-      [object | _tail] -> {:reply, object, state}
-    end
-  end
-
-  def handle_call({:get_store_pid, store, session}, _from, state) do
-    :ets.match(@registry, {store, :"$1", session, :_, :_})
-    |> case do
-      [] -> {:reply, :undefined, state}
-      [[pid] | _tail] -> {:reply, pid, state}
-    end
-  end
-
-  def handle_call({:get_store_instances, query}, _from, state) do
-    instances = :ets.match_object(@registry, query)
-
-    {:reply, instances, state}
-  end
-
-  def handle_call({:session_stores, session}, _from, state) do
-    stores = :ets.match_object(@registry, {:_, :_, session, :_, :_})
-
-    {:reply, stores, state}
-  end
-
-  def handle_info({:DOWN, _ref, :process, pid, _reason}, _state) do
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
     :ets.match_delete(@registry, {:_, pid, :_, :_, :_})
 
-    {:noreply, :ok}
+    {:noreply, state}
   end
 end

lib/storex/socket.ex

@@ -21,20 +21,20 @@ defmodule Storex.Socket do
   end
 
   def message_handle(%{type: "join"} = message, state) do
-    with {:get_module, {:ok, _}} <- {:get_module, get_store_module(message.store)},
+    with {:get_module, {:ok, _}} <- {:get_module, Storex.Store.resolve(message.store)},
          {:add_store, {:ok, _}} <-
            {:add_store,
-            Storex.Supervisor.add_store(message.store, state.session, state.pid, message.data)} do
-      store_state = Storex.Supervisor.get_store_state(state.session, message.store)
-
+            Storex.Supervisor.add_store(message.store, state.session, state.pid, message.data)},
+         {:store_state, {:ok, store_state}} <-
+           {:store_state, Storex.Supervisor.get_store_state(state.session, message.store)} do
       message =
         Map.put(message, :data, store_state)
         |> Map.put(:session, state.session)
         |> Jason.encode!()
 
       {:text, message, state}
     else
-      {:add_store, {:error, error_message}} ->
+      {step, {:error, error_message}} when step in [:add_store, :store_state] ->
         %{
           type: "error",
           session: state.session,
@@ -50,10 +50,15 @@ defmodule Storex.Socket do
     end
   end
 
-  def message_handle(%{type: "mutation", session: session, store: store} = message, state) do
+  # The `session` carried by the frame is never used to resolve the store. It is
+  # supplied by the client, so trusting it would let any socket mutate any other
+  # session's store. Only `state.session`, generated by the handler for this
+  # connection, addresses a store. Mutating another session deliberately is what
+  # `Storex.mutate/3` and `Storex.mutate/4` are for.
+  def message_handle(%{type: "mutation", store: store} = message, %{session: session} = state) do
     Storex.Supervisor.mutate_store(
-      message.session,
-      message.store,
+      session,
+      store,
       message.data.name,
       message.data.data
     )
@@ -89,27 +94,4 @@ defmodule Storex.Socket do
     |> Jason.encode!()
     |> (&{:text, &1, state}).()
   end
-
-  defp safe_concat(store) do
-    try do
-      module = Module.safe_concat([store])
-      {:ok, module}
-    rescue
-      ArgumentError -> {:error, :not_exists}
-    end
-  end
-
-  defp get_store_module(store) do
-    with {:ok, module} <- safe_concat(store),
-         {:module, module} <- Code.ensure_compiled(module),
-         true <-
-           Storex.Store in (module.module_info(:attributes)
-                            |> Keyword.get_values(:behaviour)
-                            |> List.flatten()) do
-      {:ok, module}
-    else
-      false -> {:error, :not_store}
-      _ -> {:error, :not_exists}
-    end
-  end
 end