Tag mount in longpoll priv_topic

lib/phoenix/endpoint.ex

@@ -758,7 +758,12 @@ defmodule Phoenix.Endpoint do
     paths =
       if longpoll do
         longpoll = put_auth_token(longpoll, opts[:auth_token])
-        config = Phoenix.Socket.Transport.load_config(longpoll, Phoenix.Transports.LongPoll)
+
+        config =
+          longpoll
+          |> Phoenix.Socket.Transport.load_config(Phoenix.Transports.LongPoll)
+          |> Phoenix.Transports.LongPoll.put_mount_tag(socket, path)
+
         plug_init = {endpoint, socket, config}
         {conn_ast, match_path} = socket_path(path, config)
         [{match_path, Phoenix.Transports.LongPoll, conn_ast, plug_init} | paths]

lib/phoenix/transports/long_poll.ex

@@ -8,6 +8,7 @@ defmodule Phoenix.Transports.LongPoll do
   @max_poll_batch_size 100
   @connect_info_opts [:check_csrf]
   @token_header "x-phoenix-longpoll-token"
+  @topic_prefix "phx:lp:"
 
   import Plug.Conn
   alias Phoenix.Socket.{V1, V2, Transport}
@@ -25,6 +26,30 @@ defmodule Phoenix.Transports.LongPoll do
 
   def init(opts), do: opts
 
+  @doc false
+  # A session token is bound to the mount that issued it by embedding this tag
+  # in the session's private topic. The topic is opaque to the rest of the
+  # transport -- it is only ever used as a pubsub topic -- so servers running an
+  # older Phoenix keep accepting tokens that carry a tag, which makes this safe
+  # to roll out to a mixed fleet.
+  #
+  # The tag must be stable across nodes and reboots, so it is derived
+  # exclusively from the compile-time identity of the mount.
+  def put_mount_tag(config, handler, path) do
+    tag =
+      :crypto.hash(:sha256, [
+        Atom.to_string(handler),
+        0,
+        path,
+        0,
+        Keyword.fetch!(config, :path)
+      ])
+      |> binary_part(0, 9)
+      |> Base.url_encode64(padding: false)
+
+    Keyword.put(config, :mount_tag, tag)
+  end
+
   def call(conn, {endpoint, handler, opts}) do
     conn
     |> fetch_query_params()
@@ -133,7 +158,9 @@ defmodule Phoenix.Transports.LongPoll do
 
   defp new_session(conn, endpoint, handler, opts) do
     priv_topic =
-      "phx:lp:" <>
+      @topic_prefix <>
+        Keyword.fetch!(opts, :mount_tag) <>
+        ":" <>
         Base.encode64(:crypto.strong_rand_bytes(16)) <>
         (System.system_time(:millisecond) |> Integer.to_string())
 
@@ -202,7 +229,8 @@ defmodule Phoenix.Transports.LongPoll do
   # by publishing a message in the encrypted private topic.
   defp resume_session(%Plug.Conn{} = conn, endpoint, opts) do
     with token when is_binary(token) <- fetch_token(conn),
-         {:ok, {:v1, id, pid, priv_topic}} <- verify_token(endpoint, token, opts) do
+         {:ok, {:v1, id, pid, priv_topic}} <- verify_token(endpoint, token, opts),
+         :ok <- check_mount(priv_topic, opts) do
       server_ref = server_ref(endpoint.config(:endpoint_id), id, pid, priv_topic)
 
       new_conn =
@@ -227,6 +255,20 @@ defmodule Phoenix.Transports.LongPoll do
 
   ## Helpers
 
+  # Rejects a token that was issued at a different mount. The tag is separated
+  # from the random part by a ":", which the random part itself can never
+  # contain, so a topic without one is a token created before mount binding
+  # existed. Those are still accepted so that rolling deploys do not tear down
+  # live sessions; the clause can be dropped in a later release.
+  defp check_mount(@topic_prefix <> rest, opts) do
+    case :binary.split(rest, ":") do
+      [tag, _random] -> if tag == opts[:mount_tag], do: :ok, else: :error
+      [_legacy] -> :ok
+    end
+  end
+
+  defp check_mount(_priv_topic, _opts), do: :error
+
   defp server_ref(endpoint_id, id, pid, topic) when is_pid(pid) do
     cond do
       node(pid) in Node.list() -> pid

test/phoenix/integration/long_poll_socket_test.exs

@@ -73,6 +73,10 @@ defmodule Phoenix.Integration.LongPollSocketTest do
     socket "/custom/:socket_var", UserSocket,
       longpoll: [path: ":path_var/path", check_origin: ["//example.com"], pubsub_timeout_ms: 200],
       custom: :value
+
+    socket "/shortlived", UserSocket,
+      longpoll: [window_ms: 200, pubsub_timeout_ms: 200, crypto: [max_age: 1]],
+      custom: :value
   end
 
   setup %{adapter: adapter} do
@@ -108,6 +112,24 @@ defmodule Phoenix.Integration.LongPollSocketTest do
     do_poll(method, path, params, body, headers)
   end
 
+  # Re-signs a live session with the private topic format used before session
+  # tokens were bound to their mount. The session is reached through the pid in
+  # the token -- the topic is only consulted when the owning node is remote --
+  # so the rewritten topic still resolves to the same session.
+  defp downgrade_token(token) do
+    salt = Atom.to_string(Endpoint.config(:pubsub_server))
+
+    {:ok, {:v1, id, pid, _topic}} =
+      Phoenix.Token.verify(Endpoint, salt, token, max_age: 1_209_600)
+
+    legacy_topic =
+      "phx:lp:" <>
+        Base.encode64(:crypto.strong_rand_bytes(16)) <>
+        (System.system_time(:millisecond) |> Integer.to_string())
+
+    Phoenix.Token.sign(Endpoint, salt, {:v1, id, pid, legacy_topic})
+  end
+
   defp do_poll(method, path, params, body, headers) do
     headers = Map.merge(%{"content-type" => "application/json"}, headers)
     url = "http://127.0.0.1:#{@port}/#{path}?" <> URI.encode_query(params)
@@ -206,6 +228,60 @@ defmodule Phoenix.Integration.LongPollSocketTest do
         resp = poll(:get, "ws/longpoll", secret, nil)
         assert resp.body["messages"] == ["pong"]
       end
+
+      test "binds the session token to the mount that issued it" do
+        resp = poll(:get, "ws/longpoll", %{"hello" => "world"}, nil)
+        secret = Map.take(resp.body, ["token"])
+
+        resp = poll(:post, "ws/longpoll", secret, "ping")
+        assert resp.body["status"] == 200
+
+        # the very same token is not accepted at a sibling mount
+        resp = poll(:post, "custom/123/456/path", secret, "ping")
+        assert resp.body["status"] == 410
+
+        # and a poll there starts a fresh session instead of resuming it
+        resp = poll(:get, "custom/123/456/path", secret, nil)
+        assert resp.body["status"] == 410
+        assert resp.body["messages"] == []
+        assert resp.body["token"] != secret["token"]
+
+        # the original session is left untouched throughout
+        resp = poll(:get, "ws/longpoll", secret, nil)
+        assert resp.body["messages"] == ["pong"]
+      end
+
+      test "accepts legacy tokens that carry no mount tag" do
+        resp = poll(:get, "ws/longpoll", %{"hello" => "world"}, nil)
+        legacy = %{"token" => downgrade_token(resp.body["token"])}
+
+        resp = poll(:post, "ws/longpoll", legacy, "params")
+        assert resp.body["status"] == 200
+
+        resp = poll(:get, "ws/longpoll", legacy, nil)
+        assert resp.body["messages"] == [~s(%{"hello" => "world"})]
+      end
+
+      test "does not let a sibling mount extend a token past its max_age" do
+        resp = poll(:get, "shortlived/longpoll", %{}, nil)
+        salt = Atom.to_string(Endpoint.config(:pubsub_server))
+
+        {:ok, payload} = Phoenix.Token.verify(Endpoint, salt, resp.body["token"], max_age: 1)
+
+        # the same session, signed as if it had been issued two minutes ago
+        stale = %{
+          "token" =>
+            Phoenix.Token.sign(Endpoint, salt, payload, signed_at: System.os_time(:second) - 120)
+        }
+
+        # the issuing mount rejects it, since its max_age is one second
+        resp = poll(:post, "shortlived/longpoll", stale, "ping")
+        assert resp.body["status"] == 410
+
+        # and it cannot be laundered through a mount with a longer max_age
+        resp = poll(:post, "ws/longpoll", stale, "ping")
+        assert resp.body["status"] == 410
+      end
     end
   end
 end