Merge branch 'release/2.4.4' into mergeback/2.4.4

This commit is contained in:
Tusooa Zhu 2022-10-08 22:15:09 -04:00
commit dd82fd234f
No known key found for this signature in database
GPG Key ID: 7B467EDE43A08224
9 changed files with 172 additions and 10 deletions

View File

@ -56,6 +56,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

View File

@ -94,7 +94,8 @@ def start(_type, _args) 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) ++

View File

@ -32,7 +32,8 @@ def init(%{qs: qs} = req, state) 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} ->
@ -52,7 +53,7 @@ def websocket_init(state) do
"#{__MODULE__} accepted websocket connection for user #{(state.user || %{id: "anonymous"}).id}, topic #{state.topic}"
)
Streamer.add_socket(state.topic, state.user)
Streamer.add_socket(state.topic, state.oauth_token)
{:ok, %{state | timer: timer()}}
end
@ -98,6 +99,10 @@ def websocket_info(:tick, state) 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

View File

@ -21,6 +21,18 @@ def revoke(%App{} = app, %{"token" => token} = _attrs) 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

View File

@ -37,7 +37,7 @@ def registry, do: @registry
{: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 @@ def get_topic(_stream, _user, _oauth_token, _params) 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}
@ -338,6 +338,22 @@ defp thread_containment(activity, user) 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.

View File

@ -4,7 +4,7 @@ defmodule Pleroma.Mixfile do
def project do
[
app: :pleroma,
version: version("2.4.52"),
version: version("2.4.53"),
elixir: "~> 1.10",
elixirc_paths: elixirc_paths(Mix.env()),
compilers: [:phoenix, :gettext] ++ Mix.compilers(),

View File

@ -93,7 +93,7 @@ test "receives well formatted events" 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
@ -130,5 +130,21 @@ test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
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

View File

@ -887,4 +887,105 @@ test "it sends conversation update to the 'direct' stream when a message is dele
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

View File

@ -41,6 +41,12 @@ def handle_frame(frame, state) do
{:ok, state}
end
@impl true
def handle_disconnect(conn_status, state) do
send(state.sender, {:close, conn_status})
{:ok, state}
end
@doc false
@impl true
def handle_info({:text, msg}, state) do