Page MenuHomePhorge
Files F85827588

No OneTemporary
Actions

Authored By
Unknown
Size
7 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/changelog.d/publisher-worker-atoms.fix b/changelog.d/publisher-worker-atoms.fix
new file mode 100644
index 000000000..2df6608ba
--- /dev/null
+++ b/changelog.d/publisher-worker-atoms.fix
@@ -0,0 +1 @@
+Avoid creating atoms from serialized PublisherWorker parameters.
diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex
index f799af77a..4ac0e218e 100644
--- a/lib/pleroma/workers/publisher_worker.ex
+++ b/lib/pleroma/workers/publisher_worker.ex
@@ -1,58 +1,71 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.PublisherWorker do
alias Pleroma.Activity
alias Pleroma.Instances
alias Pleroma.Web.Federator
use Oban.Worker, queue: :federator_outgoing, max_attempts: 13
+ @publish_one_param_keys %{
+ "activity_id" => :activity_id,
+ "cc" => :cc,
+ "inbox" => :inbox,
+ "unreachable_since" => :unreachable_since
+ }
+
@impl true
def perform(%Job{args: %{"op" => "publish", "activity_id" => activity_id}}) do
activity = Activity.get_by_id(activity_id)
Federator.perform(:publish, activity)
end
def perform(%Job{args: %{"op" => "publish_one", "params" => params}} = job) do
- params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
+ params =
+ Enum.reduce(params, %{}, fn {key, value}, acc ->
+ case @publish_one_param_keys do
+ %{^key => atom_key} -> Map.put(acc, atom_key, value)
+ _ -> acc
+ end
+ end)
# Cancel / skip the job if this server believed to be unreachable now
if not Instances.reachable?(params.inbox) do
{:cancel, :unreachable}
else
case Federator.perform(:publish_one, params) do
{:ok, _} ->
:ok
{:error, _} = error ->
# Only mark as unreachable on final failure
if job.attempt == job.max_attempts do
Instances.set_unreachable(params.inbox)
end
error
error ->
# Unexpected error, may have been client side
error
end
end
end
@impl true
def timeout(_job), do: :timer.seconds(10)
@base_backoff 15
@pow 5
@impl true
def backoff(%Job{attempt: attempt}) when is_integer(attempt) do
backoff =
:math.pow(attempt, @pow) +
@base_backoff +
:rand.uniform(2 * @base_backoff) * attempt
trunc(backoff)
end
end
diff --git a/test/pleroma/workers/publisher_worker_test.exs b/test/pleroma/workers/publisher_worker_test.exs
index 4519864a5..0c71415fd 100644
--- a/test/pleroma/workers/publisher_worker_test.exs
+++ b/test/pleroma/workers/publisher_worker_test.exs
@@ -1,123 +1,149 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.PublisherWorkerTest do
use Pleroma.DataCase, async: false
use Oban.Testing, repo: Pleroma.Repo
import Pleroma.Factory
import Mock
alias Pleroma.Instances
alias Pleroma.Object
+ alias Pleroma.Workers.PublisherWorker
alias Pleroma.Web.ActivityPub.ActivityPub
alias Pleroma.Web.ActivityPub.Builder
alias Pleroma.Web.CommonAPI
alias Pleroma.Web.Federator
describe "Oban job priority:" do
setup do
user = insert(:user)
{:ok, post} = CommonAPI.post(user, %{status: "Regrettable post"})
object = Object.normalize(post, fetch: false)
{:ok, delete_data, _meta} = Builder.delete(user, object.data["id"])
{:ok, delete, _meta} = ActivityPub.persist(delete_data, local: true)
%{
post: post,
delete: delete
}
end
test "Deletions are lower priority", %{delete: delete} do
assert {:ok, %Oban.Job{priority: 3}} = Federator.publish(delete)
end
test "Creates are normal priority", %{post: post} do
assert {:ok, %Oban.Job{priority: 0}} = Federator.publish(post)
end
end
describe "Server reachability:" do
setup do
user = insert(:user)
remote_user = insert(:user, local: false, inbox: "https://example.com/inbox")
{:ok, _, _} = Pleroma.User.follow(remote_user, user)
{:ok, activity} = CommonAPI.post(user, %{status: "Test post"})
%{
user: user,
remote_user: remote_user,
activity: activity
}
end
test "marks server as unreachable only on final failure", %{activity: activity} do
with_mock Pleroma.Web.Federator,
perform: fn :publish_one, _params -> {:error, :connection_error} end do
# First attempt
job = %Oban.Job{
args: %{
"op" => "publish_one",
"params" => %{
"inbox" => "https://example.com/inbox",
"activity_id" => activity.id
}
},
attempt: 1,
max_attempts: 5
}
- assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job)
+ assert {:error, :connection_error} = PublisherWorker.perform(job)
assert Instances.reachable?("https://example.com/inbox")
# Final attempt
job = %{job | attempt: 5}
- assert {:error, :connection_error} = Pleroma.Workers.PublisherWorker.perform(job)
+ assert {:error, :connection_error} = PublisherWorker.perform(job)
refute Instances.reachable?("https://example.com/inbox")
end
end
test "does not mark server as unreachable on successful publish", %{activity: activity} do
with_mock Pleroma.Web.Federator,
perform: fn :publish_one, _params -> {:ok, %{status: 200}} end do
job = %Oban.Job{
args: %{
"op" => "publish_one",
"params" => %{
"inbox" => "https://example.com/inbox",
"activity_id" => activity.id
}
},
attempt: 1,
max_attempts: 5
}
- assert :ok = Pleroma.Workers.PublisherWorker.perform(job)
+ assert :ok = PublisherWorker.perform(job)
assert Instances.reachable?("https://example.com/inbox")
end
end
+ test "does not create atoms from unexpected job params", %{activity: activity} do
+ unknown_key = "unknown_#{System.unique_integer([:positive])}"
+
+ assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end
+
+ with_mock Pleroma.Web.Federator,
+ perform: fn :publish_one, _params -> {:ok, %{status: 200}} end do
+ job = %Oban.Job{
+ args: %{
+ "op" => "publish_one",
+ "params" => %{
+ "inbox" => "https://example.com/inbox",
+ "activity_id" => activity.id,
+ unknown_key => "ignored"
+ }
+ },
+ attempt: 1,
+ max_attempts: 5
+ }
+
+ assert :ok = PublisherWorker.perform(job)
+ assert_raise ArgumentError, fn -> String.to_existing_atom(unknown_key) end
+ end
+ end
+
test "cancels job if server is unreachable", %{activity: activity} do
# First mark the server as unreachable
Instances.set_unreachable("https://example.com/inbox")
refute Instances.reachable?("https://example.com/inbox")
job = %Oban.Job{
args: %{
"op" => "publish_one",
"params" => %{
"inbox" => "https://example.com/inbox",
"activity_id" => activity.id
}
},
attempt: 1,
max_attempts: 5
}
- assert {:cancel, :unreachable} = Pleroma.Workers.PublisherWorker.perform(job)
+ assert {:cancel, :unreachable} = PublisherWorker.perform(job)
end
end
end

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 30, 9:42 PM (16 h, 22 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1800578
Default Alt Text
(7 KB)

Event Timeline