No OneTemporary
Actions

Authored By

Unknown

Size

23 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/streamer/worker.ex b/lib/pleroma/web/streamer/worker.ex
	index c2ee9e1f5..33b24840d 100644
	--- a/lib/pleroma/web/streamer/worker.ex
	+++ b/lib/pleroma/web/streamer/worker.ex
	@@ -1,224 +1,224 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer.Worker do
	use GenServer

	require Logger

	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer.State
	alias Pleroma.Web.Streamer.StreamerSocket
	alias Pleroma.Web.StreamerView

	def start_link(_) do
	GenServer.start_link(__MODULE__, %{}, [])
	end

	def init(init_arg) do
	{:ok, init_arg}
	end

	def stream(pid, topics, items) do
	GenServer.call(pid, {:stream, topics, items})
	end

	def handle_call({:stream, topics, item}, _from, state) when is_list(topics) do
	Enum.each(topics, fn t ->
	do_stream(%{topic: t, item: item})
	end)

	{:reply, state, state}
	end

	def handle_call({:stream, topic, items}, _from, state) when is_list(items) do
	Enum.each(items, fn i ->
	do_stream(%{topic: topic, item: i})
	end)

	{:reply, state, state}
	end

	def handle_call({:stream, topic, item}, _from, state) do
	do_stream(%{topic: topic, item: item})

	{:reply, state, state}
	end

	defp do_stream(%{topic: "direct", item: item}) do
	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "direct:#{id}" end)

	Enum.each(recipient_topics, fn user_topic ->
	Logger.debug("Trying to push direct message to #{user_topic}\n\n")
	push_to_socket(State.get_sockets(), user_topic, item)
	end)
	end

	defp do_stream(%{topic: "participation", item: participation}) do
	user_topic = "direct:#{participation.user_id}"
	Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

	push_to_socket(State.get_sockets(), user_topic, participation)
	end

	defp do_stream(%{topic: "list", item: item}) do
	# filter the recipient list if the activity is not public, see #270.
	recipient_lists =
	case Visibility.is_public?(item) do
	true ->
	Pleroma.List.get_lists_from_activity(item)

	_ ->
	Pleroma.List.get_lists_from_activity(item)
	\|> Enum.filter(fn list ->
	owner = User.get_cached_by_id(list.user_id)

	Visibility.visible_for_user?(item, owner)
	end)
	end

	recipient_topics =
	recipient_lists
	\|> Enum.map(fn %{id: id} -> "list:#{id}" end)

	Enum.each(recipient_topics, fn list_topic ->
	Logger.debug("Trying to push message to #{list_topic}\n\n")
	push_to_socket(State.get_sockets(), list_topic, item)
	end)
	end

	defp do_stream(%{topic: topic, item: %Notification{} = item})
	when topic in ["user", "user:notification"] do
	State.get_sockets()
	\|> Map.get("#{topic}:#{item.user_id}", [])
	\|> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} ->
	with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id),
	true <- should_send?(user, item) do
	send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)})
	end
	end)
	end

	defp do_stream(%{topic: "user", item: item}) do
	Logger.debug("Trying to push to users")

	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "user:#{id}" end)

	Enum.each(recipient_topics, fn topic ->
	push_to_socket(State.get_sockets(), topic, item)
	end)
	end

	defp do_stream(%{topic: topic, item: item}) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(State.get_sockets(), topic, item)
	end

	defp should_send?(%User{} = user, %Activity{} = item) do
	blocks = user.blocks \|\| []
	mutes = user.mutes \|\| []
	reblog_mutes = user.muted_reblogs \|\| []
	recipient_blocks = MapSet.new(blocks ++ mutes)
	recipients = MapSet.new(item.recipients)
	domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)

	- with parent when not is_nil(parent) <- Object.normalize(item),
	+ with parent <- Object.normalize(item) \|\| item,
	true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
	true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
	true <- MapSet.disjoint?(recipients, recipient_blocks),
	%{host: item_host} <- URI.parse(item.actor),
	%{host: parent_host} <- URI.parse(parent.data["actor"]),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
	true <- thread_containment(item, user),
	false <- CommonAPI.thread_muted?(user, item) do
	true
	else
	_ -> false
	end
	end

	defp should_send?(%User{} = user, %Notification{activity: activity}) do
	should_send?(user, activity)
	end

	def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	transport_pid: transport_pid,
	user: socket_user
	} ->
	# Get the current user so we have up-to-date blocks etc.
	if socket_user do
	user = User.get_cached_by_ap_id(socket_user.ap_id)

	if should_send?(user, item) do
	send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
	end
	else
	send(transport_pid, {:text, StreamerView.render("update.json", item)})
	end
	end)
	end

	def push_to_socket(topics, topic, %Participation{} = participation) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	send(transport_pid, {:text, StreamerView.render("conversation.json", participation)})
	end)
	end

	def push_to_socket(topics, topic, %Activity{
	data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	}) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	send(
	transport_pid,
	{:text, %{event: "delete", payload: to_string(deleted_activity_id)} \|> Jason.encode!()}
	)
	end)
	end

	def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop

	def push_to_socket(topics, topic, item) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	transport_pid: transport_pid,
	user: socket_user
	} ->
	# Get the current user so we have up-to-date blocks etc.
	if socket_user do
	user = User.get_cached_by_ap_id(socket_user.ap_id)

	if should_send?(user, item) do
	send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
	end
	else
	send(transport_pid, {:text, StreamerView.render("update.json", item)})
	end
	end)
	end

	@spec thread_containment(Activity.t(), User.t()) :: boolean()
	defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true

	defp thread_containment(activity, user) do
	if Config.get([:instance, :skip_thread_containment]) do
	true
	else
	ActivityPub.contain_activity(activity, user)
	end
	end
	end
	diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
	index cb1015171..d6968f4d0 100644
	--- a/test/web/streamer/streamer_test.exs
	+++ b/test/web/streamer/streamer_test.exs
	@@ -1,556 +1,574 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerTest do
	use Pleroma.DataCase

	import Pleroma.Factory

	alias Pleroma.List
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.Streamer.StreamerSocket
	alias Pleroma.Web.Streamer.Worker

	@moduletag needs_streamer: true
	clear_config_all([:instance, :skip_thread_containment])

	describe "user streams" do
	setup do
	user = insert(:user)
	notify = insert(:notification, user: user, activity: build(:note_activity))
	{:ok, %{user: user, notify: notify}}
	end

	test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

	Streamer.add_socket(
	"user",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user", notify)
	Task.await(task)
	end

	test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

	Streamer.add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user:notification", notify)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
	user: user
	} do
	blocked = insert(:user)
	{:ok, user} = User.block(user, blocked)

	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

	Streamer.add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
	{:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)

	Streamer.stream("user:notification", notif)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
	user: user
	} do
	user2 = insert(:user)
	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

	Streamer.add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	{:ok, activity} = CommonAPI.add_mute(user, activity)
	{:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
	Streamer.stream("user:notification", notif)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
	user: user
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

	Streamer.add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	{:ok, notif, _} = CommonAPI.favorite(activity.id, user2)

	Streamer.stream("user:notification", notif)
	Task.await(task)
	end
	+
	+ test "it sends follow activities to the 'user:notification' stream", %{
	+ user: user
	+ } do
	+ user2 = insert(:user)
	+ task = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
	+
	+ Streamer.add_socket(
	+ "user:notification",
	+ %{transport_pid: task.pid, assigns: %{user: user}}
	+ )
	+
	+ {:ok, _follower, _followed, _activity} = CommonAPI.follow(user2, user)
	+
	+ # We don't directly pipe the notification to the streamer as it's already
	+ # generated as a side effect of CommonAPI.follow().
	+ Task.await(task)
	+ end
	end

	test "it sends to public" do
	user = insert(:user)
	other_user = insert(:user)

	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	{:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)

	task =
	Task.async(fn ->
	expected_event =
	%{
	"event" => "delete",
	"payload" => activity.id
	}
	\|> Jason.encode!()

	assert_receive {:text, received_event}, 4_000
	assert received_event == expected_event
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	{:ok, activity} = CommonAPI.delete(activity.id, other_user)

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	describe "thread_containment" do
	test "it doesn't send to user if recipients invalid and thread containment is enabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], false)
	author = insert(:user)
	user = insert(:user)
	User.follow(user, author, "accept")

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	test "it sends message if recipients invalid and thread containment is disabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], true)
	author = insert(:user)
	user = insert(:user)
	User.follow(user, author, "accept")

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], false)
	author = insert(:user)
	user = insert(:user, skip_thread_containment: true)
	User.follow(user, author, "accept")

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end
	end

	describe "blocks" do
	test "it doesn't send messages involving blocked users" do
	user = insert(:user)
	blocked_user = insert(:user)
	{:ok, user} = User.block(user, blocked_user)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	{:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	test "it doesn't send messages transitively involving blocked users" do
	blocker = insert(:user)
	blockee = insert(:user)
	friend = insert(:user)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: blocker
	}

	topics = %{
	"public" => [fake_socket]
	}

	{:ok, blocker} = User.block(blocker, blockee)

	{:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})

	Worker.push_to_socket(topics, "public", activity_one)

	{:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})

	Worker.push_to_socket(topics, "public", activity_two)

	{:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})

	Worker.push_to_socket(topics, "public", activity_three)

	Task.await(task)
	end
	end

	test "it doesn't send unwanted DMs to list" do
	user_a = insert(:user)
	user_b = insert(:user)
	user_c = insert(:user)

	{:ok, user_a} = User.follow(user_a, user_b)

	{:ok, list} = List.create("Test", user_a)
	{:ok, list} = List.follow(list, user_b)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	"status" => "@#{user_c.nickname} Test",
	"visibility" => "direct"
	})

	topics = %{
	"list:#{list.id}" => [fake_socket]
	}

	Worker.handle_call({:stream, "list", activity}, self(), topics)

	Task.await(task)
	end

	test "it doesn't send unwanted private posts to list" do
	user_a = insert(:user)
	user_b = insert(:user)

	{:ok, list} = List.create("Test", user_a)
	{:ok, list} = List.follow(list, user_b)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	"status" => "Test",
	"visibility" => "private"
	})

	topics = %{
	"list:#{list.id}" => [fake_socket]
	}

	Worker.handle_call({:stream, "list", activity}, self(), topics)

	Task.await(task)
	end

	test "it sends wanted private posts to list" do
	user_a = insert(:user)
	user_b = insert(:user)

	{:ok, user_a} = User.follow(user_a, user_b)

	{:ok, list} = List.create("Test", user_a)
	{:ok, list} = List.follow(list, user_b)

	task =
	Task.async(fn ->
	assert_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	"status" => "Test",
	"visibility" => "private"
	})

	Streamer.add_socket(
	"list:#{list.id}",
	fake_socket
	)

	Worker.handle_call({:stream, "list", activity}, self(), %{})

	Task.await(task)
	end

	test "it doesn't send muted reblogs" do
	user1 = insert(:user)
	user2 = insert(:user)
	user3 = insert(:user)
	CommonAPI.hide_reblogs(user1, user2)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user1
	}

	{:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
	{:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", announce_activity)

	Task.await(task)
	end

	test "it doesn't send posts from muted threads" do
	user = insert(:user)
	user2 = insert(:user)
	{:ok, user2, user, _activity} = CommonAPI.follow(user2, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})

	{:ok, activity} = CommonAPI.add_mute(user2, activity)

	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

	Process.sleep(4000)

	Streamer.add_socket(
	"user",
	%{transport_pid: task.pid, assigns: %{user: user2}}
	)

	Streamer.stream("user", activity)
	Task.await(task)
	end

	describe "direct streams" do
	setup do
	:ok
	end

	test "it sends conversation update to the 'direct' stream", %{} do
	user = insert(:user)
	another_user = insert(:user)

	task =
	Task.async(fn ->
	assert_receive {:text, _received_event}, 4_000
	end)

	Streamer.add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hey @#{user.nickname}",
	"visibility" => "direct"
	})

	Task.await(task)
	end

	test "it doesn't send conversation update to the 'direct' streamj when the last message in the conversation is deleted" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, 4_000
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	refute_receive {:text, _}, 4_000
	end)

	Process.sleep(1000)

	Streamer.add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _} = CommonAPI.delete(create_activity.id, another_user)

	Task.await(task)
	end

	test "it sends conversation update to the 'direct' stream when a message is deleted" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	{:ok, create_activity2} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"in_reply_to_status_id" => create_activity.id,
	"visibility" => "direct"
	})

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, 4_000
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	assert_receive {:text, received_event}, 4_000

	assert %{"event" => "conversation", "payload" => received_payload} =
	Jason.decode!(received_event)

	assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	assert last_status["id"] == to_string(create_activity.id)
	end)

	Process.sleep(1000)

	Streamer.add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _} = CommonAPI.delete(create_activity2.id, another_user)

	Task.await(task)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sun, Nov 1, 1:19 AM (11 h, 47 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1802058
Default Alt Text: (23 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions