No OneTemporary
Actions

Size

22 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index aa75cb728..9c3956683 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,297 +1,297 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Delivery
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.FedSockets

	require Pleroma.Constants

	import Pleroma.Web.ActivityPub.Visibility

	@behaviour Pleroma.Web.Federator.Publisher

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def is_representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	Publish a single message to a peer. Takes a struct with the following
	parameters set:

	* `inbox`: the inbox to publish to
	* `json`: the JSON message body representing the ActivityPub message
	* `actor`: the actor which is signing the message
	* `id`: the ActivityStreams URI of the message
	"""
	def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
	Logger.debug("Federating #{id} to #{inbox}")

	case FedSockets.get_or_create_fed_socket(inbox) do
	{:ok, fedsocket} ->
	Logger.debug("publishing via fedsockets - #{inspect(inbox)}")
	FedSockets.publish(fedsocket, json)

	_ ->
	Logger.debug("publishing via http - #{inspect(inbox)}")
	http_publish(inbox, actor, json, params)
	end
	end

	def publish_one(%{actor_id: actor_id} = params) do
	actor = User.get_cached_by_id(actor_id)

	params
	\|> Map.delete(:actor_id)
	\|> Map.put(:actor, actor)
	\|> publish_one()
	end

	defp http_publish(inbox, actor, json, params) do
	- %{host: host, path: path} = URI.parse(inbox)
	+ uri = %{path: path} = URI.parse(inbox)
	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date = Pleroma.Signature.signed_date()

	signature =
	Pleroma.Signature.sign(actor, %{
	- "(request-target)": "post #{uri.path}",
	+ "(request-target)": "post #{path}",
	host: signature_host(uri),
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	with {:ok, %{status: code}} when code in 200..299 <-
	result =
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	if not Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since] do
	Instances.set_reachable(inbox)
	end

	result
	else
	{_post_result, response} ->
	unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	{:error, response}
	end
	end

	defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
	if port == URI.default_port(scheme) do
	host
	else
	"#{host}:#{port}"
	end
	end

	defp should_federate?(inbox, public) do
	if public do
	true
	else
	%{host: host} = URI.parse(inbox)

	quarantined_instances =
	Config.get([:instance, :quarantined_instances], [])
	\|> Pleroma.Web.ActivityPub.MRF.subdomains_regex()

	!Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
	end
	end

	@spec recipients(User.t(), Activity.t()) :: list(User.t()) \| []
	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	User.get_external_followers(actor)
	else
	[]
	end

	fetchers =
	with %Activity{data: %{"type" => "Delete"}} <- activity,
	%Object{id: object_id} <- Object.normalize(activity),
	fetchers <- User.get_delivered_users_by_object_id(object_id),
	_ <- Delivery.delete_all_by_object_id(object_id) do
	fetchers
	else
	_ ->
	[]
	end

	Pleroma.Web.Federator.Publisher.remote_users(actor, activity) ++ followers ++ fetchers
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	defp maybe_use_sharedinbox(%User{shared_inbox: nil, inbox: inbox}), do: inbox
	defp maybe_use_sharedinbox(%User{shared_inbox: shared_inbox}), do: shared_inbox

	@doc """
	Determine a user inbox to use based on heuristics. These heuristics
	are based on an approximation of the ``sharedInbox`` rules in the
	[ActivityPub specification][ap-sharedinbox].

	Please do not edit this function (or its children) without reading
	the spec, as editing the code is likely to introduce some breakage
	without some familiarity.

	[ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
	"""
	def determine_inbox(
	%Activity{data: activity_data},
	%User{inbox: inbox} = user
	) do
	to = activity_data["to"] \|\| []
	cc = activity_data["cc"] \|\| []
	type = activity_data["type"]

	cond do
	type == "Delete" ->
	maybe_use_sharedinbox(user)

	Pleroma.Constants.as_public() in to \|\| Pleroma.Constants.as_public() in cc ->
	maybe_use_sharedinbox(user)

	length(to) + length(cc) > 1 ->
	maybe_use_sharedinbox(user)

	true ->
	inbox
	end
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
	when is_list(bcc) and bcc != [] do
	public = is_public?(activity)
	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)

	recipients = recipients(actor, activity)

	inboxes =
	recipients
	\|> Enum.filter(&User.ap_enabled?/1)
	\|> Enum.map(fn actor -> actor.inbox end)
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()

	Repo.checkout(fn ->
	Enum.each(inboxes, fn {inbox, unreachable_since} ->
	%User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# instance would only accept a first message for the first recipient and ignore the rest.
	cc = get_cc_ap_ids(ap_id, recipients)

	json =
	data
	\|> Map.put("cc", cc)
	\|> Jason.encode!()

	Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	})
	end)
	end)
	end

	@doc """
	Publishes an activity to all relevant peers.
	"""
	def publish(%User{} = actor, %Activity{} = activity) do
	public = is_public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
	Relay.publish(activity)
	end

	{:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
	json = Jason.encode!(data)

	recipients(actor, activity)
	\|> Enum.filter(fn user -> User.ap_enabled?(user) end)
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
	\|> Instances.filter_reachable()
	\|> Enum.each(fn {inbox, unreachable_since} ->
	Pleroma.Web.Federator.Publisher.enqueue_one(
	__MODULE__,
	%{
	inbox: inbox,
	json: json,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	}
	)
	end)
	end

	def gather_webfinger_links(%User{} = user) do
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.base_url()}/ostatus_subscribe?acct={uri}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/fed_sockets/fed_sockets.ex b/lib/pleroma/web/fed_sockets/fed_sockets.ex
	index 9685e3efe..51a7b0b99 100644
	--- a/lib/pleroma/web/fed_sockets/fed_sockets.ex
	+++ b/lib/pleroma/web/fed_sockets/fed_sockets.ex
	@@ -1,182 +1,182 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets do
	@moduledoc """
	This documents the FedSockets framework. A framework for transferring
	ActivityPub objects between servers via persistant WebSocket connections.

	FedSockets allow servers to authenticate on first contact and maintain that
	connection, eliminating the need to authenticate every time data needs to move.

	## Protocol
	FedSockets currently support 2 types of data transfer:
	* `publish` method which doesn't require a response
	* `fetch` method requires a response be sent

	### Publish
	The publish operation sends a json encoded map of the shape:
	%{action: :publish, data: json}
	and accepts (but does not require) a reply of form:
	%{"action" => "publish_reply"}

	The outgoing params represent
	* data: ActivityPub object encoded into json


	### Fetch
	The fetch operation sends a json encoded map of the shape:
	%{action: :fetch, data: id, uuid: fetch_uuid}
	and requires a reply of form:
	%{"action" => "fetch_reply", "uuid" => uuid, "data" => data}

	The outgoing params represent
	* id: an ActivityPub object URI
	* uuid: a unique uuid generated by the sender

	The reply params represent
	* data: an ActivityPub object encoded into json
	* uuid: the uuid sent along with the fetch request

	## Examples
	Clients of FedSocket transfers shouldn't need to use any of the functions outside of this module.

	A typical publish operation can be performed through the following code, and a fetch operation in a similar manner.

	case FedSockets.get_or_create_fed_socket(inbox) do
	{:ok, fedsocket} ->
	FedSockets.publish(fedsocket, json)

	_ ->
	alternative_publish(inbox, actor, json, params)
	end

	## Configuration
	FedSockets have the following config settings

	config :pleroma, :fed_sockets,
	enabled: true,
	ping_interval: :timer.seconds(15),
	connection_duration: :timer.hours(1),
	rejection_duration: :timer.hours(1),
	fed_socket_fetches: [
	default: 12_000,
	interval: 3_000,
	lazy: false
	]
	* enabled - turn FedSockets on or off with this flag. Can be toggled at runtime.
	* connection_duration - How long a FedSocket can sit idle before it's culled.
	* rejection_duration - After failing to make a FedSocket connection a host will be excluded
	from further connections for this amount of time
	* fed_socket_fetches - Use these parameters to pass options to the Cachex queue backing the FetchRegistry
	* fed_socket_rejections - Use these parameters to pass options to the Cachex queue backing the FedRegistry

	Cachex options are
	* default: the minimum amount of time a fetch can wait before it times out.
	* interval: the interval between checks for timed out entries. This plus the default represent the maximum time allowed
	* lazy: leave at false for consistant and fast lookups, set to true for stricter timeout enforcement

	"""
	require Logger

	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.SocketInfo

	@doc """
	returns a FedSocket for the given origin. Will reuse an existing one or create a new one.

	address is expected to be a fully formed URL such as:
	"http://www.example.com" or "http://www.example.com:8080"

	It can and usually does include additional path parameters,
	but these are ignored as the FedSockets are organized by host and port info alone.
	"""
	def get_or_create_fed_socket(address) do
	with {:cache, {:error, :missing}} <- {:cache, get_fed_socket(address)},
	{:connect, {:ok, _pid}} <- {:connect, FedSocket.connect_to_host(address)},
	{:cache, {:ok, fed_socket}} <- {:cache, get_fed_socket(address)} do
	Logger.debug("fedsocket created for - #{inspect(address)}")
	{:ok, fed_socket}
	else
	{:cache, {:ok, socket}} ->
	Logger.debug("fedsocket found in cache - #{inspect(address)}")
	{:ok, socket}

	{:connect, {:error, _host}} ->
	Logger.debug("set host rejected for - #{inspect(address)}")
	FedRegistry.set_host_rejected(address)
	{:error, :rejected}

	{_, {:error, :disabled}} ->
	{:error, :disabled}

	{_, {:error, reason}} ->
	Logger.warn("get_or_create_fed_socket error - #{inspect(reason)}")
	{:error, reason}
	end
	end

	@doc """
	returns a FedSocket for the given origin. Will not create a new FedSocket if one does not exist.

	address is expected to be a fully formed URL such as:
	"http://www.example.com" or "http://www.example.com:8080"
	"""
	def get_fed_socket(address) do
	origin = SocketInfo.origin(address)

	with {:config, true} <- {:config, Pleroma.Config.get([:fed_sockets, :enabled], false)},
	{:ok, socket} <- FedRegistry.get_fed_socket(origin) do
	{:ok, socket}
	else
	{:config, _} ->
	{:error, :disabled}

	{:error, :rejected} ->
	- Logger.error("FedSocket previously rejected - #{inspect(origin)}")
	+ Logger.debug("FedSocket previously rejected - #{inspect(origin)}")
	{:error, :rejected}

	{:error, reason} ->
	{:error, reason}
	end
	end

	@doc """
	Sends the supplied data via the publish protocol.
	It will not block waiting for a reply.
	Returns :ok but this is not an indication of a successful transfer.

	the data is expected to be JSON encoded binary data.
	"""
	def publish(%SocketInfo{} = fed_socket, json) do
	FedSocket.publish(fed_socket, json)
	end

	@doc """
	Sends the supplied data via the fetch protocol.
	It will block waiting for a reply or timeout.

	Returns {:ok, object} where object is the requested object (or nil)
	{:error, :timeout} in the event the message was not responded to

	the id is expected to be the URI of an ActivityPub object.
	"""
	def fetch(%SocketInfo{} = fed_socket, id) do
	FedSocket.fetch(fed_socket, id)
	end

	@doc """
	Disconnect all and restart FedSockets.
	This is mainly used in development and testing but could be useful in production.
	"""
	def reset do
	FedRegistry
	\|> Process.whereis()
	\|> Process.exit(:testing)
	end

	def uri_for_origin(origin),
	do: "ws://#{origin}/api/fedsocket/v1"
	end
	diff --git a/lib/pleroma/web/fed_sockets/incoming_handler.ex b/lib/pleroma/web/fed_sockets/incoming_handler.ex
	index 589ed0da0..49d0d9d84 100644
	--- a/lib/pleroma/web/fed_sockets/incoming_handler.ex
	+++ b/lib/pleroma/web/fed_sockets/incoming_handler.ex
	@@ -1,88 +1,88 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets.IncomingHandler do
	require Logger

	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.SocketInfo

	import HTTPSignatures, only: [validate_conn: 1, split_signature: 1]

	@behaviour :cowboy_websocket

	def init(req, state) do
	shake = FedSocket.shake()

	with true <- Pleroma.Config.get([:fed_sockets, :enabled]),
	sec_protocol <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	headers = %{"(request-target)" => ^shake} <- :cowboy_req.headers(req),
	true <- validate_conn(%{req_headers: headers}),
	%{"keyId" => origin} <- split_signature(headers["signature"]) do
	req =
	if is_nil(sec_protocol) do
	req
	else
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_protocol, req)
	end

	{:cowboy_websocket, req, %{origin: origin}, %{}}
	else
	_ ->
	{:ok, req, state}
	end
	end

	def websocket_init(%{origin: origin}) do
	case FedRegistry.add_fed_socket(origin) do
	{:ok, socket_info} ->
	{:ok, socket_info}

	e ->
	Logger.error("FedSocket websocket_init failed - #{inspect(e)}")
	{:error, inspect(e)}
	end
	end

	# Use the ping to check if the connection should be expired
	def websocket_handle(:ping, socket_info) do
	if SocketInfo.expired?(socket_info) do
	{:stop, socket_info}
	else
	- {:ok, socket_info}
	+ {:ok, socket_info, :hibernate}
	end
	end

	def websocket_handle({:text, data}, socket_info) do
	socket_info = SocketInfo.touch(socket_info)

	case FedSocket.receive_package(socket_info, data) do
	{:noreply, _} ->
	{:ok, socket_info}

	{:reply, reply} ->
	{:reply, {:text, Jason.encode!(reply)}, socket_info}

	{:error, reason} ->
	Logger.error("incoming error - receive_package: #{inspect(reason)}")
	{:ok, socket_info}
	end
	end

	def websocket_info({:send, message}, socket_info) do
	socket_info = SocketInfo.touch(socket_info)

	{:reply, {:text, message}, socket_info}
	end

	def websocket_info(:close, state) do
	{:stop, state}
	end

	def websocket_info(message, state) do
	Logger.debug("#{__MODULE__} unknown message #{inspect(message)}")
	{:ok, state}
	end
	end
	diff --git a/lib/pleroma/web/fed_sockets/outgoing_handler.ex b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
	index d859966f0..88a8c25dc 100644
	--- a/lib/pleroma/web/fed_sockets/outgoing_handler.ex
	+++ b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
	@@ -1,127 +1,131 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.FedSockets.OutgoingHandler do
	use GenServer

	require Logger

	alias Pleroma.Web.ActivityPub.InternalFetchActor
	alias Pleroma.Web.FedSockets
	alias Pleroma.Web.FedSockets.FedRegistry
	alias Pleroma.Web.FedSockets.FedSocket
	alias Pleroma.Web.FedSockets.SocketInfo

	def start_link(origin) do
	GenServer.start_link(__MODULE__, %{origin: origin})
	end

	def init(%{origin: origin}) do
	uri_string = FedSockets.uri_for_origin(origin)

	case initiate_connection(uri_string) do
	{:ok, conn_pid} ->
	FedRegistry.add_fed_socket(uri_string, conn_pid)

	{:error, reason} ->
	Logger.debug("Outgoing connection failed - #{inspect(reason)}")
	:ignore
	end
	end

	def handle_info({:gun_ws, conn_pid, _ref, {:text, data}}, socket_info) do
	socket_info = SocketInfo.touch(socket_info)

	case FedSocket.receive_package(socket_info, data) do
	{:noreply, _} ->
	{:noreply, socket_info}

	{:reply, reply} ->
	:gun.ws_send(conn_pid, {:text, Jason.encode!(reply)})
	{:noreply, socket_info}

	{:error, reason} ->
	Logger.error("incoming error - receive_package: #{inspect(reason)}")
	{:noreply, socket_info}
	end
	end

	def handle_info(:close, state) do
	Logger.debug("Sending close frame !!!!!!!")
	{:close, state}
	end

	def handle_info({:gun_down, _pid, _prot, :closed, _}, state) do
	{:stop, :normal, state}
	end

	def handle_info({:send, data}, %{conn_pid: conn_pid} = socket_info) do
	socket_info = SocketInfo.touch(socket_info)
	:gun.ws_send(conn_pid, {:text, data})
	{:noreply, socket_info}
	end

	+ def handle_info({:gun_ws, _, _, :pong}, state) do
	+ {:noreply, state, :hibernate}
	+ end
	+
	def handle_info(msg, state) do
	Logger.debug("#{__MODULE__} unhandled event #{inspect(msg)}")
	{:noreply, state}
	end

	def terminate(reason, state) do
	Logger.debug(
	"#{__MODULE__} terminating outgoing connection for #{inspect(state)} for #{inspect(reason)}"
	)

	{:ok, state}
	end

	def initiate_connection(uri_string) do
	uri = %{host: host, port: port, path: path} = URI.parse(uri_string)

	with {:ok, conn_pid} <- :gun.open(to_charlist(host), port),
	{:ok, _} <- :gun.await_up(conn_pid),
	headers <- build_headers(uri),
	- ref <- :gun.ws_upgrade(conn_pid, to_charlist(path), headers) do
	+ ref <- :gun.ws_upgrade(conn_pid, to_charlist(path), headers, %{silence_pings: false}) do
	receive do
	{:gun_upgrade, ^conn_pid, ^ref, [<<"websocket">>], _} ->
	{:ok, conn_pid}
	after
	15_000 ->
	Logger.debug("Fedsocket timeout connecting to #{inspect(uri)}")
	{:error, :timeout}
	end
	else
	e ->
	Logger.debug("Fedsocket error connecting to #{inspect(uri)}")
	{:error, e}
	end
	end

	defp build_headers(%{host: host, port: nil}), do: build_headers("#{host}")
	defp build_headers(%{host: host, port: port}), do: build_headers("#{host}:#{port}")

	defp build_headers(host) when is_binary(host) do
	shake = FedSocket.shake()
	digest = "SHA-256=" <> (:crypto.hash(:sha256, shake) \|> Base.encode64())
	date = Pleroma.Signature.signed_date()
	shake_size = byte_size(shake)

	signature_opts = %{
	"(request-target)": shake,
	"content-length": to_charlist("#{shake_size}"),
	date: date,
	digest: digest,
	host: host
	}

	signature = Pleroma.Signature.sign(InternalFetchActor.get_actor(), signature_opts)

	[
	{'signature', to_charlist(signature)},
	{'date', date},
	{'digest', to_charlist(digest)},
	{'content-length', to_charlist("#{shake_size}")},
	{to_charlist("(request-target)"), to_charlist(shake)}
	]
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sun, Oct 11, 6:49 AM (1 d, 12 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1785639
Default Alt Text: (22 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions