summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--CHANGELOG.md5
-rw-r--r--lib/pleroma/application.ex3
-rw-r--r--lib/pleroma/web/mastodon_api/websocket_handler.ex9
-rw-r--r--lib/pleroma/web/o_auth/token/strategy/revoke.ex14
-rw-r--r--lib/pleroma/web/streamer.ex24
-rw-r--r--mix.exs4
-rw-r--r--mix.lock2
-rw-r--r--test/pleroma/integration/mastodon_websocket_test.exs36
-rw-r--r--test/pleroma/web/streamer_test.exs101
-rw-r--r--test/support/websocket_client.ex32
10 files changed, 197 insertions, 33 deletions
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 95405bb60..bcbe3ba56 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
### Removed
+## 2.4.4 - 2022-08-19
+
+### Security
+- Streaming API sessions will now properly disconnect if the corresponding token is revoked
+
## 2.4.3 - 2022-05-06
### Security
diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 9824e0a4a..92d143665 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -89,7 +89,8 @@ defmodule Pleroma.Application do
Pleroma.Repo,
Config.TransferTask,
Pleroma.Emoji,
- Pleroma.Web.Plugs.RateLimiter.Supervisor
+ Pleroma.Web.Plugs.RateLimiter.Supervisor,
+ {Task.Supervisor, name: Pleroma.TaskSupervisor}
] ++
cachex_children() ++
http_children(adapter, @mix_env) ++
diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
index 0d1faffbd..930e9eb29 100644
--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
@@ -32,7 +32,8 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
req
end
- {:cowboy_websocket, req, %{user: user, topic: topic, count: 0, timer: nil},
+ {:cowboy_websocket, req,
+ %{user: user, topic: topic, oauth_token: oauth_token, count: 0, timer: nil},
%{idle_timeout: @timeout}}
else
{:error, :bad_topic} ->
@@ -54,7 +55,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
}, topic #{state.topic}"
)
- Streamer.add_socket(state.topic, state.user)
+ Streamer.add_socket(state.topic, state.oauth_token)
{:ok, %{state | timer: timer()}}
end
@@ -100,6 +101,10 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
{:reply, :ping, %{state | timer: nil, count: 0}, :hibernate}
end
+ def websocket_info(:close, state) do
+ {:stop, state}
+ end
+
# State can be `[]` only in case we terminate before switching to websocket,
# we already log errors for these cases in `init/1`, so just do nothing here
def terminate(_reason, _req, []), do: :ok
diff --git a/lib/pleroma/web/o_auth/token/strategy/revoke.ex b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
index 8d6572704..de99bc137 100644
--- a/lib/pleroma/web/o_auth/token/strategy/revoke.ex
+++ b/lib/pleroma/web/o_auth/token/strategy/revoke.ex
@@ -21,6 +21,18 @@ defmodule Pleroma.Web.OAuth.Token.Strategy.Revoke do
@doc "Revokes access token"
@spec revoke(Token.t()) :: {:ok, Token.t()} | {:error, Ecto.Changeset.t()}
def revoke(%Token{} = token) do
- Repo.delete(token)
+ with {:ok, token} <- Repo.delete(token) do
+ Task.Supervisor.start_child(
+ Pleroma.TaskSupervisor,
+ Pleroma.Web.Streamer,
+ :close_streams_by_oauth_token,
+ [token],
+ restart: :transient
+ )
+
+ {:ok, token}
+ else
+ result -> result
+ end
end
end
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
index fc3bbb130..8bf70d99b 100644
--- a/lib/pleroma/web/streamer.ex
+++ b/lib/pleroma/web/streamer.ex
@@ -37,7 +37,7 @@ defmodule Pleroma.Web.Streamer do
{:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
with {:ok, topic} <- get_topic(stream, user, oauth_token, params) do
- add_socket(topic, user)
+ add_socket(topic, oauth_token)
end
end
@@ -120,10 +120,10 @@ defmodule Pleroma.Web.Streamer do
end
@doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
- def add_socket(topic, user) do
+ def add_socket(topic, oauth_token) do
if should_env_send?() do
- auth? = if user, do: true
- Registry.register(@registry, topic, auth?)
+ oauth_token_id = if oauth_token, do: oauth_token.id, else: false
+ Registry.register(@registry, topic, oauth_token_id)
end
{:ok, topic}
@@ -320,6 +320,22 @@ defmodule Pleroma.Web.Streamer do
end
end
+ def close_streams_by_oauth_token(oauth_token) do
+ if should_env_send?() do
+ Registry.select(
+ @registry,
+ [
+ {
+ {:"$1", :"$2", :"$3"},
+ [{:==, :"$3", oauth_token.id}],
+ [:"$2"]
+ }
+ ]
+ )
+ |> Enum.each(fn pid -> send(pid, :close) end)
+ end
+ end
+
# In test environement, only return true if the registry is started.
# In benchmark environment, returns false.
# In any other environment, always returns true.
diff --git a/mix.exs b/mix.exs
index 927f39975..0e2834fc6 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Pleroma.Mixfile do
def project do
[
app: :pleroma,
- version: version("2.4.3"),
+ version: version("2.4.4"),
elixir: "~> 1.9",
elixirc_paths: elixirc_paths(Mix.env()),
compilers: [:phoenix, :gettext] ++ Mix.compilers(),
@@ -210,7 +210,7 @@ defmodule Pleroma.Mixfile do
{:excoveralls, "0.12.3", only: :test},
{:hackney, "~> 1.18.0", override: true},
{:mox, "~> 1.0", only: :test},
- {:websocket_client, git: "https://github.com/jeremyong/websocket_client.git", only: :test}
+ {:websockex, "~> 0.4.3", only: :test}
] ++ oauth_deps()
end
diff --git a/mix.lock b/mix.lock
index 821c397b4..1fe713e8e 100644
--- a/mix.lock
+++ b/mix.lock
@@ -126,5 +126,5 @@
"unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"},
"unsafe": {:hex, :unsafe, "1.0.1", "a27e1874f72ee49312e0a9ec2e0b27924214a05e3ddac90e91727bc76f8613d8", [:mix], [], "hexpm", "6c7729a2d214806450d29766abc2afaa7a2cbecf415be64f36a6691afebb50e5"},
"web_push_encryption": {:git, "https://github.com/lanodan/elixir-web-push-encryption.git", "026a043037a89db4da8f07560bc8f9c68bcf0cc0", [branch: "bugfix/otp-24"]},
- "websocket_client": {:git, "https://github.com/jeremyong/websocket_client.git", "9a6f65d05ebf2725d62fb19262b21f1805a59fbf", []},
+ "websockex": {:hex, :websockex, "0.4.3", "92b7905769c79c6480c02daacaca2ddd49de936d912976a4d3c923723b647bf0", [:mix], [], "hexpm", "95f2e7072b85a3a4cc385602d42115b73ce0b74a9121d0d6dbbf557645ac53e4"},
}
diff --git a/test/pleroma/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
index 43ec57893..d44033842 100644
--- a/test/pleroma/integration/mastodon_websocket_test.exs
+++ b/test/pleroma/integration/mastodon_websocket_test.exs
@@ -33,16 +33,18 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
test "refuses invalid requests" do
capture_log(fn ->
- assert {:error, {404, _}} = start_socket()
- assert {:error, {404, _}} = start_socket("?stream=ncjdk")
+ assert {:error, %WebSockex.RequestError{code: 404}} = start_socket()
+ assert {:error, %WebSockex.RequestError{code: 404}} = start_socket("?stream=ncjdk")
Process.sleep(30)
end)
end
test "requires authentication and a valid token for protected streams" do
capture_log(fn ->
- assert {:error, {401, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
- assert {:error, {401, _}} = start_socket("?stream=user")
+ assert {:error, %WebSockex.RequestError{code: 401}} =
+ start_socket("?stream=user&access_token=aaaaaaaaaaaa")
+
+ assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
Process.sleep(30)
end)
end
@@ -91,7 +93,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
{:ok, token} = OAuth.Token.exchange_token(app, auth)
- %{user: user, token: token}
+ %{app: app, user: user, token: token}
end
test "accepts valid tokens", state do
@@ -102,7 +104,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
capture_log(fn ->
- assert {:error, {401, _}} = start_socket("?stream=user")
+ assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
Process.sleep(30)
end)
end
@@ -111,7 +113,9 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
capture_log(fn ->
- assert {:error, {401, _}} = start_socket("?stream=user:notification")
+ assert {:error, %WebSockex.RequestError{code: 401}} =
+ start_socket("?stream=user:notification")
+
Process.sleep(30)
end)
end
@@ -120,11 +124,27 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])
capture_log(fn ->
- assert {:error, {401, _}} =
+ assert {:error, %WebSockex.RequestError{code: 401}} =
start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])
Process.sleep(30)
end)
end
+
+ test "disconnect when token is revoked", %{app: app, user: user, token: token} do
+ assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
+ assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
+
+ {:ok, auth} = OAuth.Authorization.create_authorization(app, user)
+
+ {:ok, token2} = OAuth.Token.exchange_token(app, auth)
+ assert {:ok, _} = start_socket("?stream=user&access_token=#{token2.token}")
+
+ OAuth.Token.Strategy.Revoke.revoke(token)
+
+ assert_receive {:close, _}
+ assert_receive {:close, _}
+ refute_receive {:close, _}
+ end
end
end
diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs
index b788a9138..7c4b9e288 100644
--- a/test/pleroma/web/streamer_test.exs
+++ b/test/pleroma/web/streamer_test.exs
@@ -813,4 +813,105 @@ defmodule Pleroma.Web.StreamerTest do
assert last_status["id"] == to_string(create_activity.id)
end
end
+
+ describe "stop streaming if token got revoked" do
+ setup do
+ child_proc = fn start, finalize ->
+ fn ->
+ start.()
+
+ receive do
+ {StreamerTest, :ready} ->
+ assert_receive {:render_with_user, _, "update.json", _}
+
+ receive do
+ {StreamerTest, :revoked} -> finalize.()
+ end
+ end
+ end
+ end
+
+ starter = fn user, token ->
+ fn -> Streamer.get_topic_and_add_socket("user", user, token) end
+ end
+
+ hit = fn -> assert_receive :close end
+ miss = fn -> refute_receive :close end
+
+ send_all = fn tasks, thing -> Enum.each(tasks, &send(&1.pid, thing)) end
+
+ %{
+ child_proc: child_proc,
+ starter: starter,
+ hit: hit,
+ miss: miss,
+ send_all: send_all
+ }
+ end
+
+ test "do not revoke other tokens", %{
+ child_proc: child_proc,
+ starter: starter,
+ hit: hit,
+ miss: miss,
+ send_all: send_all
+ } do
+ %{user: user, token: token} = oauth_access(["read"])
+ %{token: token2} = oauth_access(["read"], user: user)
+ %{user: user2, token: user2_token} = oauth_access(["read"])
+
+ post_user = insert(:user)
+ CommonAPI.follow(user, post_user)
+ CommonAPI.follow(user2, post_user)
+
+ tasks = [
+ Task.async(child_proc.(starter.(user, token), hit)),
+ Task.async(child_proc.(starter.(user, token2), miss)),
+ Task.async(child_proc.(starter.(user2, user2_token), miss))
+ ]
+
+ {:ok, _} =
+ CommonAPI.post(post_user, %{
+ status: "hi"
+ })
+
+ send_all.(tasks, {StreamerTest, :ready})
+
+ Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+ send_all.(tasks, {StreamerTest, :revoked})
+
+ Enum.each(tasks, &Task.await/1)
+ end
+
+ test "revoke all streams for this token", %{
+ child_proc: child_proc,
+ starter: starter,
+ hit: hit,
+ send_all: send_all
+ } do
+ %{user: user, token: token} = oauth_access(["read"])
+
+ post_user = insert(:user)
+ CommonAPI.follow(user, post_user)
+
+ tasks = [
+ Task.async(child_proc.(starter.(user, token), hit)),
+ Task.async(child_proc.(starter.(user, token), hit))
+ ]
+
+ {:ok, _} =
+ CommonAPI.post(post_user, %{
+ status: "hi"
+ })
+
+ send_all.(tasks, {StreamerTest, :ready})
+
+ Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
+
+ send_all.(tasks, {StreamerTest, :revoked})
+
+ Enum.each(tasks, &Task.await/1)
+ end
+ end
end
diff --git a/test/support/websocket_client.ex b/test/support/websocket_client.ex
index 34b955474..70d331999 100644
--- a/test/support/websocket_client.ex
+++ b/test/support/websocket_client.ex
@@ -5,18 +5,17 @@
defmodule Pleroma.Integration.WebsocketClient do
# https://github.com/phoenixframework/phoenix/blob/master/test/support/websocket_client.exs
+ use WebSockex
+
@doc """
Starts the WebSocket server for given ws URL. Received Socket.Message's
are forwarded to the sender pid
"""
def start_link(sender, url, headers \\ []) do
- :crypto.start()
- :ssl.start()
-
- :websocket_client.start_link(
- String.to_charlist(url),
+ WebSockex.start_link(
+ url,
__MODULE__,
- [sender],
+ %{sender: sender},
extra_headers: headers
)
end
@@ -36,27 +35,32 @@ defmodule Pleroma.Integration.WebsocketClient do
end
@doc false
- def init([sender], _conn_state) do
- {:ok, %{sender: sender}}
+ @impl true
+ def handle_frame(frame, state) do
+ send(state.sender, frame)
+ {:ok, state}
end
- @doc false
- def websocket_handle(frame, _conn_state, state) do
- send(state.sender, frame)
+ @impl true
+ def handle_disconnect(conn_status, state) do
+ send(state.sender, {:close, conn_status})
{:ok, state}
end
@doc false
- def websocket_info({:text, msg}, _conn_state, state) do
+ @impl true
+ def handle_info({:text, msg}, state) do
{:reply, {:text, msg}, state}
end
- def websocket_info(:close, _conn_state, _state) do
+ @impl true
+ def handle_info(:close, _state) do
{:close, <<>>, "done"}
end
@doc false
- def websocket_terminate(_reason, _conn_state, _state) do
+ @impl true
+ def terminate(_reason, _state) do
:ok
end
end
generated by cgit v1.2.3 (git 2.25.1) at 2025-10-28 09:23:57 +0000