Page MenuHomePhorge
Files F85825477

No OneTemporary
Actions

Authored By
Unknown
Size
15 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lib/pleroma/web/federator.ex b/lib/pleroma/web/federator.ex
index e812b1a46..58260afa8 100644
--- a/lib/pleroma/web/federator.ex
+++ b/lib/pleroma/web/federator.ex
@@ -1,130 +1,131 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.Federator do
alias Pleroma.Activity
alias Pleroma.Object.Containment
alias Pleroma.User
alias Pleroma.Web.ActivityPub.Publisher
alias Pleroma.Web.ActivityPub.Transmogrifier
alias Pleroma.Web.ActivityPub.Utils
alias Pleroma.Workers.PublisherWorker
alias Pleroma.Workers.ReceiverWorker
require Logger
@behaviour Pleroma.Web.Federator.Publishing
@doc """
Returns `true` if the distance to target object does not exceed max configured value.
Serves to prevent fetching of very long threads, especially useful on smaller instances.
Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161).
Applies to fetching of both ancestor (reply-to) and child (reply) objects.
"""
# credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
def allowed_thread_distance?(distance) do
max_distance = Pleroma.Config.get([:instance, :federation_incoming_replies_max_depth])
if max_distance && max_distance >= 0 do
# Default depth is 0 (an object has zero distance from itself in its thread)
(distance || 0) <= max_distance
else
true
end
end
# Client API
def incoming_ap_doc(%{params: params, req_headers: req_headers}) do
ReceiverWorker.new(
%{
"op" => "incoming_ap_doc",
"req_headers" => req_headers,
"params" => params,
"timeout" => :timer.seconds(20)
},
priority: 2
)
|> Oban.insert()
end
def incoming_ap_doc(%{"type" => "Delete"} = params) do
ReceiverWorker.new(%{"op" => "incoming_ap_doc", "params" => params},
priority: 3,
queue: :slow
)
|> Oban.insert()
end
def incoming_ap_doc(params) do
ReceiverWorker.new(%{"op" => "incoming_ap_doc", "params" => params})
|> Oban.insert()
end
@impl true
def publish(%{id: "pleroma:fakeid"} = activity) do
perform(:publish, activity)
end
@impl true
def publish(%Pleroma.Activity{data: %{"type" => type}} = activity) do
PublisherWorker.new(%{"op" => "publish", "activity_id" => activity.id},
priority: publish_priority(type)
)
|> Oban.insert()
end
defp publish_priority("Delete"), do: 3
defp publish_priority(_), do: 0
# Job Worker Callbacks
@spec perform(atom(), any()) :: {:ok, any()} | {:error, any()}
def perform(:publish_one, params) do
Publisher.prepare_one(params)
|> Publisher.publish_one()
end
def perform(:publish, activity) do
Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
%User{} = actor = User.get_cached_by_ap_id(activity.data["actor"])
Publisher.publish(actor, activity)
end
def perform(:incoming_ap_doc, params) do
Logger.debug("Handling incoming AP activity")
actor =
params
|> Map.get("actor")
|> Utils.get_ap_id()
# NOTE: we use the actor ID to do the containment, this is fine because an
# actor shouldn't be acting on objects outside their own AP server.
- with {_, {:ok, _user}} <- {:actor, User.get_or_fetch_by_ap_id(actor)},
+ with {_, {:ok, user}} <- {:actor, User.get_or_fetch_by_ap_id(actor)},
+ {:user_active, true} <- {:user_active, match?(true, user.is_active)},
nil <- Activity.normalize(params["id"]),
{_, :ok} <-
{:correct_origin?, Containment.contain_origin_from_id(actor, params)},
{:ok, activity} <- Transmogrifier.handle_incoming(params) do
{:ok, activity}
else
{:correct_origin?, _} ->
Logger.debug("Origin containment failure for #{params["id"]}")
{:error, :origin_containment_failed}
%Activity{} ->
Logger.debug("Already had #{params["id"]}")
{:error, :already_present}
{:actor, e} ->
Logger.debug("Unhandled actor #{actor}, #{inspect(e)}")
{:error, e}
e ->
# Just drop those for now
Logger.debug(fn -> "Unhandled activity\n" <> Jason.encode!(params, pretty: true) end)
{:error, e}
end
end
end
diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
index 6787a59ef..0373ec15f 100644
--- a/lib/pleroma/workers/receiver_worker.ex
+++ b/lib/pleroma/workers/receiver_worker.ex
@@ -1,87 +1,86 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.ReceiverWorker do
alias Pleroma.Signature
alias Pleroma.User
alias Pleroma.Web.Federator
use Oban.Worker, queue: :federator_incoming, max_attempts: 5
@impl true
def perform(%Job{
args: %{
"op" => "incoming_ap_doc",
"method" => method,
"params" => params,
"req_headers" => req_headers,
"request_path" => request_path,
"query_string" => query_string
}
}) do
# Oban's serialization converts our tuple headers to lists.
# Revert it for the signature validation.
req_headers = Enum.into(req_headers, [], &List.to_tuple(&1))
conn_data = %Plug.Conn{
method: method,
params: params,
req_headers: req_headers,
request_path: request_path,
query_string: query_string
}
- with {:ok, %User{} = actor} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]),
- {:user_active, true} <- {:user_active, match?(true, actor.is_active)},
+ with {:ok, %User{}} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]),
{:ok, _public_key} <- Signature.refetch_public_key(conn_data),
{:signature, true} <- {:signature, Signature.validate_signature(conn_data)},
{:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
{:ok, res}
else
e -> process_errors(e)
end
end
def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do
with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
{:ok, res}
else
e -> process_errors(e)
end
end
@impl true
def timeout(%_{args: %{"timeout" => timeout}}), do: timeout
def timeout(_job), do: :timer.seconds(5)
defp process_errors({:error, {:error, _} = error}), do: process_errors(error)
defp process_errors(errors) do
case errors do
# User fetch failures
{:error, :not_found} = reason -> {:cancel, reason}
{:error, :forbidden} = reason -> {:cancel, reason}
# Inactive user
- {:user_active, false} = reason -> {:cancel, reason}
+ {:error, {:user_active, false} = reason} -> {:cancel, reason}
# Validator will error and return a changeset error
# e.g., duplicate activities or if the object was deleted
{:error, {:validate, {:error, _changeset} = reason}} -> {:cancel, reason}
# Duplicate detection during Normalization
{:error, :already_present} -> {:cancel, :already_present}
# MRFs will return a reject
{:error, {:reject, _} = reason} -> {:cancel, reason}
# HTTP Sigs
{:signature, false} -> {:cancel, :invalid_signature}
# Origin / URL validation failed somewhere possibly due to spoofing
{:error, :origin_containment_failed} -> {:cancel, :origin_containment_failed}
# Unclear if this can be reached
{:error, {:side_effects, {:error, :no_object_actor}} = reason} -> {:cancel, reason}
# Catchall
{:error, _} = e -> e
e -> {:error, e}
end
end
end
diff --git a/test/pleroma/workers/receiver_worker_test.exs b/test/pleroma/workers/receiver_worker_test.exs
index 779e83eaa..4d53c44ed 100644
--- a/test/pleroma/workers/receiver_worker_test.exs
+++ b/test/pleroma/workers/receiver_worker_test.exs
@@ -1,248 +1,246 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.ReceiverWorkerTest do
use Pleroma.DataCase
use Oban.Testing, repo: Pleroma.Repo
import Mock
import Pleroma.Factory
alias Pleroma.User
alias Pleroma.Web.Federator
alias Pleroma.Workers.ReceiverWorker
test "it does not retry MRF reject" do
params = insert(:note).data
with_mock Pleroma.Web.ActivityPub.Transmogrifier,
handle_incoming: fn _ -> {:reject, "MRF"} end do
assert {:cancel, {:reject, "MRF"}} =
ReceiverWorker.perform(%Oban.Job{
args: %{"op" => "incoming_ap_doc", "params" => params}
})
end
end
test "it does not retry ObjectValidator reject" do
params =
insert(:note_activity).data
|> Map.put("id", Pleroma.Web.ActivityPub.Utils.generate_activity_id())
|> Map.put("object", %{
"type" => "Note",
"id" => Pleroma.Web.ActivityPub.Utils.generate_object_id()
})
with_mock Pleroma.Web.ActivityPub.ObjectValidator, [:passthrough],
validate: fn _, _ -> {:error, %Ecto.Changeset{}} end do
assert {:cancel, {:error, %Ecto.Changeset{}}} =
ReceiverWorker.perform(%Oban.Job{
args: %{"op" => "incoming_ap_doc", "params" => params}
})
end
end
test "it does not retry duplicates" do
params = insert(:note_activity).data
assert {:cancel, :already_present} =
ReceiverWorker.perform(%Oban.Job{
args: %{"op" => "incoming_ap_doc", "params" => params}
})
end
describe "cancels on a failed user fetch" do
setup do
Tesla.Mock.mock(fn
%{url: "https://springfield.social/users/bart"} ->
%Tesla.Env{
status: 403,
body: ""
}
%{url: "https://springfield.social/users/troymcclure"} ->
%Tesla.Env{
status: 404,
body: ""
}
%{url: "https://springfield.social/users/hankscorpio"} ->
%Tesla.Env{
status: 410,
body: ""
}
end)
end
test "when request returns a 403" do
params =
insert(:note_activity).data
|> Map.put("actor", "https://springfield.social/users/bart")
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: [],
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:cancel, {:error, :forbidden}} = ReceiverWorker.perform(oban_job)
end
test "when request returns a 404" do
params =
insert(:note_activity).data
|> Map.put("actor", "https://springfield.social/users/troymcclure")
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: [],
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:cancel, {:error, :not_found}} = ReceiverWorker.perform(oban_job)
end
test "when request returns a 410" do
params =
insert(:note_activity).data
|> Map.put("actor", "https://springfield.social/users/hankscorpio")
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: [],
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:cancel, {:error, :not_found}} = ReceiverWorker.perform(oban_job)
end
test "when user account is disabled" do
user = insert(:user)
fake_activity = URI.parse(user.ap_id) |> Map.put(:path, "/fake-activity") |> to_string
params =
insert(:note_activity, user: user).data
|> Map.put("id", fake_activity)
{:ok, %User{}} = User.set_activation(user, false)
{:ok, oban_job} =
- ReceiverWorker.new(%{
- "op" => "incoming_ap_doc",
- "method" => "POST",
- "req_headers" => [],
- "request_path" => "/inbox",
- "params" => params,
- "query_string" => ""
+ Federator.incoming_ap_doc(%{
+ method: "POST",
+ req_headers: [],
+ request_path: "/inbox",
+ params: params,
+ query_string: ""
})
- |> Oban.insert()
assert {:cancel, {:user_active, false}} = ReceiverWorker.perform(oban_job)
end
end
test "it can validate the signature" do
Tesla.Mock.mock(fn
%{url: "https://phpc.social/users/denniskoch"} ->
%Tesla.Env{
status: 200,
body: File.read!("test/fixtures/denniskoch.json"),
headers: [{"content-type", "application/activity+json"}]
}
%{url: "https://phpc.social/users/denniskoch/collections/featured"} ->
%Tesla.Env{
status: 200,
headers: [{"content-type", "application/activity+json"}],
body:
File.read!("test/fixtures/users_mock/masto_featured.json")
|> String.replace("{{domain}}", "phpc.social")
|> String.replace("{{nickname}}", "denniskoch")
}
end)
params =
File.read!("test/fixtures/receiver_worker_signature_activity.json") |> Jason.decode!()
req_headers = [
["accept-encoding", "gzip"],
["content-length", "5184"],
["content-type", "application/activity+json"],
["date", "Thu, 25 Jul 2024 13:33:31 GMT"],
["digest", "SHA-256=ouge/6HP2/QryG6F3JNtZ6vzs/hSwMk67xdxe87eH7A="],
["host", "bikeshed.party"],
[
"signature",
"keyId=\"https://mastodon.social/users/bastianallgeier#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) host date digest content-type\",signature=\"ymE3vn5Iw50N6ukSp8oIuXJB5SBjGAGjBasdTDvn+ahZIzq2SIJfmVCsIIzyqIROnhWyQoTbavTclVojEqdaeOx+Ejz2wBnRBmhz5oemJLk4RnnCH0lwMWyzeY98YAvxi9Rq57Gojuv/1lBqyGa+rDzynyJpAMyFk17XIZpjMKuTNMCbjMDy76ILHqArykAIL/v1zxkgwxY/+ELzxqMpNqtZ+kQ29znNMUBB3eVZ/mNAHAz6o33Y9VKxM2jw+08vtuIZOusXyiHbRiaj2g5HtN2WBUw1MzzfRfHF2/yy7rcipobeoyk5RvP5SyHV3WrIeZ3iyoNfmv33y8fxllF0EA==\""
],
[
"user-agent",
"http.rb/5.2.0 (Mastodon/4.3.0-nightly.2024-07-25; +https://mastodon.social/)"
]
]
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: req_headers,
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:ok, %Pleroma.Activity{}} = ReceiverWorker.perform(oban_job)
end
test "cancels due to origin containment" do
params =
insert(:note_activity).data
|> Map.put("id", "https://notorigindomain.com/activity")
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: [],
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:cancel, :origin_containment_failed} = ReceiverWorker.perform(oban_job)
end
test "canceled due to deleted object" do
params =
insert(:announce_activity).data
|> Map.put("object", "http://localhost:4001/deleted")
Tesla.Mock.mock(fn
%{url: "http://localhost:4001/deleted"} ->
%Tesla.Env{
status: 404,
body: ""
}
end)
{:ok, oban_job} =
Federator.incoming_ap_doc(%{
method: "POST",
req_headers: [],
request_path: "/inbox",
params: params,
query_string: ""
})
assert {:cancel, _} = ReceiverWorker.perform(oban_job)
end
end

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 30, 1:03 AM (1 d, 9 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1799057
Default Alt Text
(15 KB)

Event Timeline