Merge branch 'release/2.0.5' into 'stable'

Release/2.0.5

See merge request pleroma/secteam/pleroma!4
This commit is contained in:
lain 2020-05-13 08:12:09 +00:00
commit a5ccb5b0b1
38 changed files with 274 additions and 118 deletions

View file

@ -3,6 +3,21 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
## [2.0.5] - 2020-05-13
### Security
- Fix possible private status leaks in Mastodon Streaming API
### Fixed
- Crashes when trying to block a user if block federation is disabled
- Not being able to start the instance without `erlang-eldap` installed
- Users with bios over the limit getting rejected
- Follower counters not being updated on incoming follow accepts
### Upgrade notes
1. Restart Pleroma
## [2.0.4] - 2020-05-10
### Security

View file

@ -501,7 +501,15 @@ def upgrade_changeset(struct, params \\ %{}, remote? \\ false) do
params = Map.put(params, :last_refreshed_at, NaiveDateTime.utc_now())
params = if remote?, do: truncate_fields_param(params), else: params
params =
if remote? do
params
|> truncate_fields_param()
|> truncate_if_exists(:name, name_limit)
|> truncate_if_exists(:bio, bio_limit)
else
params
end
struct
|> cast(

View file

@ -604,7 +604,6 @@ def block(blocker, blocked, activity_id \\ nil, local \\ true) do
end
defp do_block(blocker, blocked, activity_id, local) do
outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
if unfollow_blocked do
@ -612,8 +611,7 @@ defp do_block(blocker, blocked, activity_id, local) do
if follow_activity, do: unfollow(blocker, blocked, nil, local)
end
with true <- outgoing_blocks,
block_data <- make_block_data(blocker, blocked, activity_id),
with block_data <- make_block_data(blocker, blocked, activity_id),
{:ok, activity} <- insert(block_data, local),
:ok <- maybe_federate(activity) do
{:ok, activity}

View file

@ -544,6 +544,9 @@ def handle_incoming(
{:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
%User{local: true} = follower <- User.get_cached_by_ap_id(follow_activity.data["actor"]),
{:ok, _relationship} <- FollowingRelationship.update(follower, followed, :follow_accept) do
User.update_follower_count(followed)
User.update_following_count(follower)
ActivityPub.accept(%{
to: follow_activity.data["to"],
type: "Accept",
@ -553,7 +556,8 @@ def handle_incoming(
activity_id: id
})
else
_e -> :error
_e ->
:error
end
end

View file

@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
alias Ecto.Changeset
alias Ecto.UUID
alias Pleroma.Activity
alias Pleroma.Config
alias Pleroma.Notification
alias Pleroma.Object
alias Pleroma.Repo
@ -169,8 +170,11 @@ def create_context(context) do
Enqueues an activity for federation if it's local
"""
@spec maybe_federate(any()) :: :ok
def maybe_federate(%Activity{local: true} = activity) do
if Pleroma.Config.get!([:instance, :federating]) do
def maybe_federate(%Activity{local: true, data: %{"type" => type}} = activity) do
outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
with true <- Config.get!([:instance, :federating]),
true <- type != "Block" || outgoing_blocks do
Pleroma.Web.Federator.publish(activity)
end

View file

@ -12,29 +12,15 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
@behaviour :cowboy_websocket
@streams [
"public",
"public:local",
"public:media",
"public:local:media",
"user",
"user:notification",
"direct",
"list",
"hashtag"
]
@anonymous_streams ["public", "public:local", "hashtag"]
# Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
@timeout :infinity
def init(%{qs: qs} = req, state) do
with params <- :cow_qs.parse_qs(qs),
with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
access_token <- List.keyfind(params, "access_token", 0),
{_, stream} <- List.keyfind(params, "stream", 0),
{:ok, user} <- allow_request(stream, [access_token, sec_websocket]),
topic when is_binary(topic) <- expand_topic(stream, params) do
access_token <- Map.get(params, "access_token"),
{:ok, user} <- authenticate_request(access_token, sec_websocket),
{:ok, topic} <- Streamer.get_topic(Map.get(params, "stream"), user, params) do
req =
if sec_websocket do
:cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
@ -44,14 +30,14 @@ def init(%{qs: qs} = req, state) do
{:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
else
{:error, code} ->
Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
{:ok, req} = :cowboy_req.reply(code, req)
{:error, :bad_topic} ->
Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
{:ok, req} = :cowboy_req.reply(404, req)
{:ok, req, state}
error ->
Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
{:ok, req} = :cowboy_req.reply(400, req)
{:error, :unauthorized} ->
Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
{:ok, req} = :cowboy_req.reply(401, req)
{:ok, req, state}
end
end
@ -93,50 +79,23 @@ def terminate(reason, _req, state) do
end
# Public streams without authentication.
defp allow_request(stream, [nil, nil]) when stream in @anonymous_streams do
defp authenticate_request(nil, nil) do
{:ok, nil}
end
# Authenticated streams.
defp allow_request(stream, [access_token, sec_websocket]) when stream in @streams do
token =
with {"access_token", token} <- access_token do
token
else
_ -> sec_websocket
end
defp authenticate_request(access_token, sec_websocket) do
token = access_token || sec_websocket
with true <- is_bitstring(token),
%Token{user_id: user_id} <- Repo.get_by(Token, token: token),
user = %User{} <- User.get_cached_by_id(user_id) do
{:ok, user}
else
_ -> {:error, 403}
_ -> {:error, :unauthorized}
end
end
# Not authenticated.
defp allow_request(stream, _) when stream in @streams, do: {:error, 403}
# No matching stream.
defp allow_request(_, _), do: {:error, 404}
defp expand_topic("hashtag", params) do
case List.keyfind(params, "tag", 0) do
{_, tag} -> "hashtag:#{tag}"
_ -> nil
end
end
defp expand_topic("list", params) do
case List.keyfind(params, "list", 0) do
{_, list} -> "list:#{list}"
_ -> nil
end
end
defp expand_topic(topic, _), do: topic
defp streamer_socket(state) do
%{transport_pid: self(), assigns: state}
end

View file

@ -36,30 +36,28 @@ def handle_call(:get_state, _from, state) do
end
def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
internal_topic = internal_topic(topic, socket)
stream_socket = StreamerSocket.from_socket(socket)
sockets_for_topic =
sockets
|> Map.get(internal_topic, [])
|> Map.get(topic, [])
|> List.insert_at(0, stream_socket)
|> Enum.uniq()
state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
state = put_in(state, [:sockets, topic], sockets_for_topic)
Logger.debug("Got new conn for #{topic}")
{:reply, state, state}
end
def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
internal_topic = internal_topic(topic, socket)
stream_socket = StreamerSocket.from_socket(socket)
sockets_for_topic =
sockets
|> Map.get(internal_topic, [])
|> Map.get(topic, [])
|> List.delete(stream_socket)
state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
state = Kernel.put_in(state, [:sockets, topic], sockets_for_topic)
{:reply, state, state}
end
@ -70,13 +68,4 @@ defp do_remove_socket(:test, _, _) do
defp do_remove_socket(_env, topic, socket) do
GenServer.call(__MODULE__, {:remove, topic, socket})
end
defp internal_topic(topic, socket)
when topic in ~w[user user:notification direct] do
"#{topic}:#{socket.assigns[:user].id}"
end
defp internal_topic(topic, _) do
topic
end
end

View file

@ -3,12 +3,77 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.Streamer do
alias Pleroma.User
alias Pleroma.Web.Streamer.State
alias Pleroma.Web.Streamer.Worker
@timeout 60_000
@mix_env Mix.env()
@public_streams ["public", "public:local", "public:media", "public:local:media"]
@user_streams ["user", "user:notification", "direct"]
@doc "Expands and authorizes a stream, and registers the process for streaming."
@spec get_topic_and_add_socket(stream :: String.t(), State.t(), Map.t() | nil) ::
{:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
def get_topic_and_add_socket(stream, socket, params \\ %{}) do
user =
case socket do
%{assigns: %{user: user}} -> user
_ -> nil
end
case get_topic(stream, user, params) do
{:ok, topic} ->
add_socket(topic, socket)
{:ok, topic}
error ->
error
end
end
@doc "Expand and authorizes a stream"
@spec get_topic(stream :: String.t(), User.t() | nil, Map.t()) ::
{:ok, topic :: String.t()} | {:error, :bad_topic}
def get_topic(stream, user, params \\ %{})
# Allow all public steams.
def get_topic(stream, _, _) when stream in @public_streams do
{:ok, stream}
end
# Allow all hashtags streams.
def get_topic("hashtag", _, %{"tag" => tag}) do
{:ok, "hashtag:" <> tag}
end
# Expand user streams.
def get_topic(stream, %User{} = user, _) when stream in @user_streams do
{:ok, stream <> ":" <> to_string(user.id)}
end
def get_topic(stream, _, _) when stream in @user_streams do
{:error, :unauthorized}
end
# List streams.
def get_topic("list", %User{} = user, %{"list" => id}) do
if Pleroma.List.get(id, user) do
{:ok, "list:" <> to_string(id)}
else
{:error, :bad_topic}
end
end
def get_topic("list", _, _) do
{:error, :unauthorized}
end
def get_topic(_, _, _) do
{:error, :bad_topic}
end
def add_socket(topic, socket) do
State.add_socket(topic, socket)
end

View file

@ -4,7 +4,7 @@ defmodule Pleroma.Mixfile do
def project do
[
app: :pleroma,
version: version("2.0.4"),
version: version("2.0.5"),
elixir: "~> 1.8",
elixirc_paths: elixirc_paths(Mix.env()),
compilers: [:phoenix, :gettext] ++ Mix.compilers(),
@ -36,7 +36,7 @@ def project do
releases: [
pleroma: [
include_executables_for: [:unix],
applications: [ex_syslogger: :load, syslog: :load],
applications: [ex_syslogger: :load, syslog: :load, eldap: :transient],
steps: [:assemble, &copy_files/1, &copy_nginx_config/1]
]
]
@ -69,8 +69,7 @@ def application do
:comeonin,
:quack,
:fast_sanitize,
:ssl,
:eldap
:ssl
],
included_applications: [:ex_syslogger]
]

View file

@ -1 +1 @@
<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1,user-scalable=no"><title>Pleroma</title><!--server-generated-meta--><link rel=icon type=image/png href=/favicon.png><link href=/static/css/vendors~app.b2603a50868c68a1c192.css rel=stylesheet><link href=/static/css/app.1055039ce3f2fe4dd110.css rel=stylesheet><link href=/static/fontello.1588431888583.css rel=stylesheet></head><body class=hidden><noscript>To use Pleroma, please enable JavaScript.</noscript><div id=app></div><script type=text/javascript src=/static/js/vendors~app.c67e1a363ece7f1f7152.js></script><script type=text/javascript src=/static/js/app.57951e6e5e198d1a1266.js></script></body></html>
<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1,user-scalable=no"><title>Pleroma</title><!--server-generated-meta--><link rel=icon type=image/png href=/favicon.png><link href=/static/css/vendors~app.b2603a50868c68a1c192.css rel=stylesheet><link href=/static/css/app.1055039ce3f2fe4dd110.css rel=stylesheet><link href=/static/fontello.1589314090288.css rel=stylesheet></head><body class=hidden><noscript>To use Pleroma, please enable JavaScript.</noscript><div id=app></div><script type=text/javascript src=/static/js/vendors~app.a516afd698489b59a809.js></script><script type=text/javascript src=/static/js/app.82334f8362acc4bbcb6f.js></script></body></html>

View file

@ -114,6 +114,8 @@
<glyph glyph-name="thumbs-up-alt" unicode="&#xf164;" d="M143 107q0 15-11 25t-25 11q-15 0-25-11t-11-25q0-15 11-25t25-11q15 0 25 11t11 25z m89 286v-357q0-15-10-25t-26-11h-160q-15 0-25 11t-11 25v357q0 14 11 25t25 10h160q15 0 26-10t10-25z m661 0q0-48-31-83 9-25 9-43 1-42-24-76 9-31 0-66-9-31-31-52 5-62-27-101-36-43-110-44h-72q-37 0-80 9t-68 16-67 22q-69 24-88 25-15 0-25 11t-11 25v357q0 14 10 25t24 11q13 1 42 33t57 67q38 49 56 67 10 10 17 27t10 27 8 34q4 22 7 34t11 29 19 28q10 11 25 11 25 0 46-6t33-15 22-22 14-25 7-28 2-25 1-22q0-21-6-43t-10-33-16-31q-1-4-5-10t-6-13-5-13h155q43 0 75-32t32-75z" horiz-adv-x="928.6" />
<glyph glyph-name="share" unicode="&#xf1e0;" d="M679 286q74 0 126-53t52-126-52-126-126-53-127 53-52 126q0 7 1 19l-201 100q-51-48-121-48-75 0-127 53t-52 126 52 126 127 53q70 0 121-48l201 100q-1 12-1 19 0 74 52 126t127 53 126-53 52-126-52-126-126-53q-71 0-122 48l-201-100q1-12 1-19t-1-19l201-100q51 48 122 48z" horiz-adv-x="857.1" />
<glyph glyph-name="binoculars" unicode="&#xf1e5;" d="M393 678v-428q0-15-11-25t-25-11v-321q0-15-10-25t-26-11h-285q-15 0-25 11t-11 25v285l139 488q4 12 17 12h237z m178 0v-392h-142v392h142z m429-500v-285q0-15-11-25t-25-11h-285q-15 0-25 11t-11 25v321q-15 0-25 11t-11 25v428h237q13 0 17-12z m-589 661v-125h-197v125q0 8 5 13t13 5h161q8 0 13-5t5-13z m375 0v-125h-197v125q0 8 5 13t13 5h161q8 0 13-5t5-13z" horiz-adv-x="1000" />
<glyph glyph-name="user-plus" unicode="&#xf234;" d="M393 357q-89 0-152 63t-62 151 62 152 152 63 151-63 63-152-63-151-151-63z m536-71h196q7 0 13-6t5-12v-107q0-8-5-13t-13-5h-196v-197q0-7-6-12t-12-6h-107q-8 0-13 6t-5 12v197h-197q-7 0-12 5t-6 13v107q0 7 6 12t12 6h197v196q0 7 5 13t13 5h107q7 0 12-5t6-13v-196z m-411-125q0-29 21-51t50-21h143v-133q-38-28-95-28h-488q-67 0-108 39t-41 106q0 30 2 58t8 61 15 60 24 55 34 45 48 30 62 11q11 0 22-10 44-34 86-51t92-17 92 17 86 51q11 10 22 10 73 0 121-54h-125q-29 0-50-21t-21-50v-107z" horiz-adv-x="1142.9" />

Before

Width:  |  Height:  |  Size: 26 KiB

After

Width:  |  Height:  |  Size: 26 KiB

Binary file not shown.

Binary file not shown.

View file

@ -346,6 +346,12 @@
"code": 59427,
"src": "fontawesome"
},
{
"uid": "4aad6bb50b02c18508aae9cbe14e784e",
"css": "share",
"code": 61920,
"src": "fontawesome"
},
{
"uid": "8b80d36d4ef43889db10bc1f0dc9a862",
"css": "user",

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View file

@ -35,7 +35,7 @@ def start_socket(qs \\ nil, headers \\ []) do
test "refuses invalid requests" do
capture_log(fn ->
assert {:error, {400, _}} = start_socket()
assert {:error, {404, _}} = start_socket()
assert {:error, {404, _}} = start_socket("?stream=ncjdk")
Process.sleep(30)
end)
@ -43,8 +43,8 @@ test "refuses invalid requests" do
test "requires authentication and a valid token for protected streams" do
capture_log(fn ->
assert {:error, {403, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
assert {:error, {403, _}} = start_socket("?stream=user")
assert {:error, {401, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
assert {:error, {401, _}} = start_socket("?stream=user")
Process.sleep(30)
end)
end
@ -103,7 +103,7 @@ test "accepts the 'user' stream", %{token: token} = _state do
assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
assert capture_log(fn ->
assert {:error, {403, "Forbidden"}} = start_socket("?stream=user")
assert {:error, {401, _}} = start_socket("?stream=user")
Process.sleep(30)
end) =~ ":badarg"
end
@ -112,7 +112,7 @@ test "accepts the 'user:notification' stream", %{token: token} = _state do
assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
assert capture_log(fn ->
assert {:error, {403, "Forbidden"}} = start_socket("?stream=user:notification")
assert {:error, {401, _}} = start_socket("?stream=user:notification")
Process.sleep(30)
end) =~ ":badarg"
end
@ -121,7 +121,7 @@ test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])
assert capture_log(fn ->
assert {:error, {403, "Forbidden"}} =
assert {:error, {401, _}} =
start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])
Process.sleep(30)

View file

@ -164,12 +164,13 @@ test "it creates a notification for user and send to the 'user' and the 'user:no
user = insert(:user)
task = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
task_user_notification = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
Streamer.add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}})
Streamer.add_socket(
"user:notification",
%{transport_pid: task_user_notification.pid, assigns: %{user: user}}
)
Streamer.get_topic_and_add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}})
Streamer.get_topic_and_add_socket("user:notification", %{
transport_pid: task_user_notification.pid,
assigns: %{user: user}
})
activity = insert(:note_activity)

View file

@ -570,7 +570,10 @@ test "returns nil for nonexistant local user" do
assert fetched_user == "not found nonexistant"
end
clear_config([:instance, :user_bio_length])
test "updates an existing user, if stale" do
Pleroma.Config.put([:instance, :user_bio_length], 1)
a_week_ago = NaiveDateTime.add(NaiveDateTime.utc_now(), -604_800)
orig_user =

View file

@ -310,7 +310,11 @@ test "cached purged after activity deletion", %{conn: conn} do
end
describe "/inbox" do
clear_config([:instance, :user_bio_length])
test "it inserts an incoming activity into the database", %{conn: conn} do
Pleroma.Config.put([:instance, :user_bio_length], 1)
data = File.read!("test/fixtures/mastodon-post-activity.json") |> Poison.decode!()
conn =

View file

@ -1351,15 +1351,44 @@ test "reverts block activity on error" do
assert Repo.aggregate(Object, :count, :id) == 0
end
clear_config([:instance, :federating])
test "creates a block activity" do
Config.put([:instance, :federating], true)
blocker = insert(:user)
blocked = insert(:user)
{:ok, activity} = ActivityPub.block(blocker, blocked)
with_mock Pleroma.Web.Federator,
publish: fn _ -> nil end do
{:ok, activity} = ActivityPub.block(blocker, blocked)
assert activity.data["type"] == "Block"
assert activity.data["actor"] == blocker.ap_id
assert activity.data["object"] == blocked.ap_id
assert activity.data["type"] == "Block"
assert activity.data["actor"] == blocker.ap_id
assert activity.data["object"] == blocked.ap_id
assert called(Pleroma.Web.Federator.publish(activity))
end
end
clear_config([:instance, :federating])
clear_config([:activitypub, :outgoing_blocks])
test "works with outgoing blocks disabled, but doesn't federate" do
Config.put([:instance, :federating], true)
Config.put([:activitypub, :outgoing_blocks], false)
blocker = insert(:user)
blocked = insert(:user)
with_mock Pleroma.Web.Federator,
publish: fn _ -> nil end do
{:ok, activity} = ActivityPub.block(blocker, blocked)
assert activity.data["type"] == "Block"
assert activity.data["actor"] == blocker.ap_id
assert activity.data["object"] == blocked.ap_id
refute called(Pleroma.Web.Federator.publish(:_))
end
end
test "reverts unblock activity on error" do

View file

@ -1120,6 +1120,12 @@ test "it works for incoming accepts which are referenced by IRI only" do
follower = User.get_cached_by_id(follower.id)
assert User.following?(follower, followed) == true
follower = User.get_by_id(follower.id)
assert follower.following_count == 1
followed = User.get_by_id(followed.id)
assert followed.follower_count == 1
end
test "it fails for incoming accepts which cannot be correlated" do

View file

@ -5,8 +5,6 @@
defmodule Pleroma.Web.MastodonAPI.SuggestionControllerTest do
use Pleroma.Web.ConnCase
alias Pleroma.Config
setup do: oauth_access(["read"])
test "returns empty result", %{conn: conn} do

View file

@ -17,11 +17,81 @@ defmodule Pleroma.Web.StreamerTest do
@moduletag needs_streamer: true, capture_log: true
@streamer_timeout 150
@streamer_timeout 300
@streamer_start_wait 10
clear_config([:instance, :skip_thread_containment])
describe "get_topic without an user" do
test "allows public" do
assert {:ok, "public"} = Streamer.get_topic("public", nil)
assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil)
assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil)
assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil)
end
test "allows hashtag streams" do
assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, %{"tag" => "cofe"})
end
test "disallows user streams" do
assert {:error, _} = Streamer.get_topic("user", nil)
assert {:error, _} = Streamer.get_topic("user:notification", nil)
assert {:error, _} = Streamer.get_topic("direct", nil)
end
test "disallows list streams" do
assert {:error, _} = Streamer.get_topic("list", nil, %{"list" => 42})
end
end
describe "get_topic with an user" do
setup do
user = insert(:user)
{:ok, %{user: user}}
end
test "allows public streams", %{user: user} do
assert {:ok, "public"} = Streamer.get_topic("public", user)
assert {:ok, "public:local"} = Streamer.get_topic("public:local", user)
assert {:ok, "public:media"} = Streamer.get_topic("public:media", user)
assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", user)
end
test "allows user streams", %{user: user} do
expected_user_topic = "user:#{user.id}"
expected_notif_topic = "user:notification:#{user.id}"
expected_direct_topic = "direct:#{user.id}"
assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user)
assert {:ok, ^expected_notif_topic} = Streamer.get_topic("user:notification", user)
assert {:ok, ^expected_direct_topic} = Streamer.get_topic("direct", user)
end
test "allows hashtag streams", %{user: user} do
assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", user, %{"tag" => "cofe"})
end
test "disallows registering to an user stream", %{user: user} do
another_user = insert(:user)
assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user)
assert {:error, _} = Streamer.get_topic("user:notification:#{another_user.id}", user)
assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user)
end
test "allows list stream that are owned by the user", %{user: user} do
{:ok, list} = List.create("Test", user)
assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
assert {:ok, _} = Streamer.get_topic("list", user, %{"list" => list.id})
end
test "disallows list stream that are not owned by the user", %{user: user} do
another_user = insert(:user)
{:ok, list} = List.create("Test", another_user)
assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
assert {:error, _} = Streamer.get_topic("list", user, %{"list" => list.id})
end
end
describe "user streams" do
setup do
user = insert(:user)
@ -35,7 +105,7 @@ test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
assert_receive {:text, _}, @streamer_timeout
end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -50,7 +120,7 @@ test "it sends notify to in the 'user:notification' stream", %{user: user, notif
assert_receive {:text, _}, @streamer_timeout
end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -70,7 +140,7 @@ test "it doesn't send notify to the 'user:notification' stream when a user is bl
task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -90,7 +160,7 @@ test "it doesn't send notify to the 'user:notification' stream when a thread is
task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -110,7 +180,7 @@ test "it doesn't send notify to the 'user:notification' stream' when a domain is
task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -127,7 +197,7 @@ test "it sends follow activities to the 'user:notification' stream", %{
Process.sleep(@streamer_start_wait)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -415,14 +485,10 @@ test "it sends wanted private posts to list" do
assert_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
transport_pid: task.pid,
user: user_a
}
Streamer.add_socket(
"list:#{list.id}",
fake_socket
Streamer.get_topic_and_add_socket(
"list",
%{transport_pid: task.pid, assigns: %{user: user_a}},
%{"list" => list.id}
)
Worker.handle_call({:stream, "list", activity}, self(), %{})
@ -497,7 +563,7 @@ test "it doesn't send posts from muted threads" do
task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"user",
%{transport_pid: task.pid, assigns: %{user: user2}}
)
@ -527,7 +593,7 @@ test "it sends conversation update to the 'direct' stream", %{} do
assert last_status["pleroma"]["direct_conversation_id"] == participation.id
end)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"direct",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -561,7 +627,7 @@ test "it doesn't send conversation update to the 'direct' stream when the last m
Process.sleep(@streamer_start_wait)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"direct",
%{transport_pid: task.pid, assigns: %{user: user}}
)
@ -604,7 +670,7 @@ test "it sends conversation update to the 'direct' stream when a message is dele
Process.sleep(@streamer_start_wait)
Streamer.add_socket(
Streamer.get_topic_and_add_socket(
"direct",
%{transport_pid: task.pid, assigns: %{user: user}}
)