Page MenuHomePhorge
Files F85828305

No OneTemporary
Actions

Authored By
Unknown
Size
20 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lib/pleroma/web/fed_sockets/fed_socket.ex b/lib/pleroma/web/fed_sockets/fed_socket.ex
index 70b1aaed1..98d64e65a 100644
--- a/lib/pleroma/web/fed_sockets/fed_socket.ex
+++ b/lib/pleroma/web/fed_sockets/fed_socket.ex
@@ -1,139 +1,137 @@
# 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.FedSocket do
@moduledoc """
The FedSocket module abstracts the actions to be taken taken on connections regardless of
whether the connection started as inbound or outbound.
Normally outside modules will have no need to call the FedSocket module directly.
"""
alias Pleroma.Object
alias Pleroma.Object.Containment
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ObjectView
alias Pleroma.Web.ActivityPub.UserView
alias Pleroma.Web.ActivityPub.Visibility
alias Pleroma.Web.FedSockets.FetchRegistry
alias Pleroma.Web.FedSockets.IngesterWorker
alias Pleroma.Web.FedSockets.OutgoingHandler
alias Pleroma.Web.FedSockets.SocketInfo
require Logger
@shake "61dd18f7-f1e6-49a4-939a-a749fcdc1103"
def connect_to_host(uri) do
- origin = SocketInfo.origin(uri)
-
- case OutgoingHandler.start_link(origin) do
+ case OutgoingHandler.start_link(uri) do
{:ok, pid} ->
{:ok, pid}
error ->
{:error, error}
end
end
def close(%SocketInfo{pid: socket_pid}),
do: Process.send(socket_pid, :close, [])
def publish(%SocketInfo{pid: socket_pid}, json) do
%{action: :publish, data: json}
|> Jason.encode!()
|> send_packet(socket_pid)
end
def fetch(%SocketInfo{pid: socket_pid}, id) do
fetch_uuid = FetchRegistry.register_fetch(id)
%{action: :fetch, data: id, uuid: fetch_uuid}
|> Jason.encode!()
|> send_packet(socket_pid)
wait_for_fetch_to_return(fetch_uuid, 0)
end
def receive_package(%SocketInfo{} = fed_socket, json) do
json
|> Jason.decode!()
|> process_package(fed_socket)
end
defp wait_for_fetch_to_return(uuid, cntr) do
case FetchRegistry.check_fetch(uuid) do
{:error, :waiting} ->
Process.sleep(:math.pow(cntr, 3) |> Kernel.trunc())
wait_for_fetch_to_return(uuid, cntr + 1)
{:error, :missing} ->
Logger.error("FedSocket fetch timed out - #{inspect(uuid)}")
{:error, :timeout}
{:ok, _fr} ->
FetchRegistry.pop_fetch(uuid)
end
end
defp process_package(%{"action" => "publish", "data" => data}, %{origin: origin} = _fed_socket) do
if Containment.contain_origin(origin, data) do
IngesterWorker.enqueue("ingest", %{"object" => data})
end
{:reply, %{"action" => "publish_reply", "status" => "processed"}}
end
defp process_package(%{"action" => "fetch_reply", "uuid" => uuid, "data" => data}, _fed_socket) do
FetchRegistry.register_fetch_received(uuid, data)
{:noreply, nil}
end
defp process_package(%{"action" => "fetch", "uuid" => uuid, "data" => ap_id}, _fed_socket) do
{:ok, data} = render_fetched_data(ap_id, uuid)
{:reply, data}
end
defp process_package(%{"action" => "publish_reply"}, _fed_socket) do
{:noreply, nil}
end
defp process_package(other, _fed_socket) do
Logger.warn("unknown json packages received #{inspect(other)}")
{:noreply, nil}
end
defp render_fetched_data(ap_id, uuid) do
{:ok,
%{
"action" => "fetch_reply",
"status" => "processed",
"uuid" => uuid,
"data" => represent_item(ap_id)
}}
end
defp represent_item(ap_id) do
case User.get_by_ap_id(ap_id) do
nil ->
object = Object.get_cached_by_ap_id(ap_id)
if Visibility.is_public?(object) do
Phoenix.View.render_to_string(ObjectView, "object.json", object: object)
else
nil
end
user ->
Phoenix.View.render_to_string(UserView, "user.json", user: user)
end
end
defp send_packet(data, socket_pid) do
Process.send(socket_pid, {:send, data}, [])
end
def shake, do: @shake
end
diff --git a/lib/pleroma/web/fed_sockets/outgoing_handler.ex b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
index 88a8c25dc..79b2696a5 100644
--- a/lib/pleroma/web/fed_sockets/outgoing_handler.ex
+++ b/lib/pleroma/web/fed_sockets/outgoing_handler.ex
@@ -1,131 +1,141 @@
# 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})
+ def start_link(uri) do
+ GenServer.start_link(__MODULE__, %{uri: uri})
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)
+ def init(%{uri: uri}) do
+ case initiate_connection(uri) do
+ {:ok, ws_origin, conn_pid} ->
+ FedRegistry.add_fed_socket(ws_origin, 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)
+ def initiate_connection(uri) do
+ ws_uri =
+ uri
+ |> SocketInfo.origin()
+ |> FedSockets.uri_for_origin()
+
+ %{host: host, port: port, path: path} = URI.parse(ws_uri)
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, %{silence_pings: false}) do
receive do
{:gun_upgrade, ^conn_pid, ^ref, [<<"websocket">>], _} ->
- {:ok, conn_pid}
+ {:ok, ws_uri, 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(uri) do
+ host_for_sig = uri |> URI.parse() |> host_signature()
- 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
+ host: host_for_sig
}
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
+
+ defp host_signature(%{host: host, scheme: scheme, port: port}) do
+ if port == URI.default_port(scheme) do
+ host
+ else
+ "#{host}:#{port}"
+ end
+ end
end
diff --git a/lib/pleroma/web/fed_sockets/socket_info.ex b/lib/pleroma/web/fed_sockets/socket_info.ex
index f00fefb12..d6fdffe1a 100644
--- a/lib/pleroma/web/fed_sockets/socket_info.ex
+++ b/lib/pleroma/web/fed_sockets/socket_info.ex
@@ -1,55 +1,52 @@
# 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.SocketInfo do
defstruct origin: nil,
pid: nil,
conn_pid: nil,
state: :default,
connected_until: nil
alias Pleroma.Web.FedSockets.SocketInfo
@default_connection_duration 15 * 60 * 1000
def build(uri, conn_pid \\ nil) do
uri
|> build_origin()
|> build_pids(conn_pid)
|> touch()
end
def touch(%SocketInfo{} = socket_info),
do: %{socket_info | connected_until: new_ttl()}
def connect(%SocketInfo{} = socket_info),
do: %{socket_info | state: :connected}
def expired?(%{connected_until: connected_until}),
do: connected_until < :erlang.monotonic_time(:millisecond)
def origin(uri),
do: build_origin(uri).origin
defp build_pids(socket_info, conn_pid),
do: struct(socket_info, pid: self(), conn_pid: conn_pid)
defp build_origin(uri) when is_binary(uri),
do: uri |> URI.parse() |> build_origin
- defp build_origin(%{host: host, port: nil}),
- do: %SocketInfo{origin: host}
-
- defp build_origin(%{host: host, port: 80}),
- do: %SocketInfo{origin: host}
+ defp build_origin(%{host: host, port: nil, scheme: scheme}),
+ do: build_origin(%{host: host, port: URI.default_port(scheme)})
defp build_origin(%{host: host, port: port}),
do: %SocketInfo{origin: "#{host}:#{port}"}
defp new_ttl do
connection_duration =
Pleroma.Config.get([:fed_sockets, :connection_duration], @default_connection_duration)
:erlang.monotonic_time(:millisecond) + connection_duration
end
end
diff --git a/test/web/fed_sockets/fed_registry_test.exs b/test/web/fed_sockets/fed_registry_test.exs
index cb5c5c329..19ac874d6 100644
--- a/test/web/fed_sockets/fed_registry_test.exs
+++ b/test/web/fed_sockets/fed_registry_test.exs
@@ -1,122 +1,124 @@
# 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.FedRegistryTest do
use ExUnit.Case
alias Pleroma.Web.FedSockets
alias Pleroma.Web.FedSockets.FedRegistry
alias Pleroma.Web.FedSockets.SocketInfo
+ @good_domain "http://good.domain"
+ @good_domain_origin "good.domain:80"
+
setup do
start_supervised({Pleroma.Web.FedSockets.Supervisor, []})
- build_test_socket("good.domain")
+ build_test_socket(@good_domain)
Process.sleep(10)
:ok
end
describe "add_fed_socket/1 without conflicting sockets" do
test "can be added" do
Process.sleep(10)
- assert {:ok, %SocketInfo{origin: origin}} = FedRegistry.get_fed_socket("good.domain")
- assert origin == "good.domain"
+ assert {:ok, %SocketInfo{origin: origin}} = FedRegistry.get_fed_socket(@good_domain_origin)
+ assert origin == "good.domain:80"
end
test "multiple origins can be added" do
- build_test_socket("anothergood.domain")
+ build_test_socket("http://anothergood.domain")
Process.sleep(10)
- assert {:ok, %SocketInfo{origin: origin_1}} = FedRegistry.get_fed_socket("good.domain")
+ assert {:ok, %SocketInfo{origin: origin_1}} =
+ FedRegistry.get_fed_socket(@good_domain_origin)
assert {:ok, %SocketInfo{origin: origin_2}} =
- FedRegistry.get_fed_socket("anothergood.domain")
+ FedRegistry.get_fed_socket("anothergood.domain:80")
- assert origin_1 == "good.domain"
- assert origin_2 == "anothergood.domain"
+ assert origin_1 == "good.domain:80"
+ assert origin_2 == "anothergood.domain:80"
assert FedRegistry.list_all() |> Enum.count() == 2
end
end
describe "add_fed_socket/1 when duplicate sockets conflict" do
setup do
- build_test_socket("good.domain")
- build_test_socket("good.domain")
+ build_test_socket(@good_domain)
+ build_test_socket(@good_domain)
Process.sleep(10)
-
:ok
end
test "will be ignored" do
assert {:ok, %SocketInfo{origin: origin, pid: pid_one}} =
- FedRegistry.get_fed_socket("good.domain")
+ FedRegistry.get_fed_socket(@good_domain_origin)
- assert origin == "good.domain"
+ assert origin == "good.domain:80"
assert FedRegistry.list_all() |> Enum.count() == 1
end
test "the newer process will be closed" do
- pid_two = build_test_socket("good.domain")
+ pid_two = build_test_socket(@good_domain)
assert {:ok, %SocketInfo{origin: origin, pid: pid_one}} =
- FedRegistry.get_fed_socket("good.domain")
+ FedRegistry.get_fed_socket(@good_domain_origin)
- assert origin == "good.domain"
+ assert origin == "good.domain:80"
Process.sleep(10)
refute Process.alive?(pid_two)
assert FedRegistry.list_all() |> Enum.count() == 1
end
end
describe "get_fed_socket/1" do
test "returns missing for unknown hosts" do
assert {:error, :missing} = FedRegistry.get_fed_socket("not_a_dmoain")
end
test "returns rejected for hosts previously rejected" do
- "rejected.domain"
+ "rejected.domain:80"
|> FedSockets.uri_for_origin()
|> FedRegistry.set_host_rejected()
- assert {:error, :rejected} = FedRegistry.get_fed_socket("rejected.domain")
+ assert {:error, :rejected} = FedRegistry.get_fed_socket("rejected.domain:80")
end
test "can retrieve a previously added SocketInfo" do
- build_test_socket("good.domain")
+ build_test_socket(@good_domain)
Process.sleep(10)
- assert {:ok, %SocketInfo{origin: origin}} = FedRegistry.get_fed_socket("good.domain")
- assert origin == "good.domain"
+ assert {:ok, %SocketInfo{origin: origin}} = FedRegistry.get_fed_socket(@good_domain_origin)
+ assert origin == "good.domain:80"
end
test "removes references to SocketInfos when the process crashes" do
assert {:ok, %SocketInfo{origin: origin, pid: pid}} =
- FedRegistry.get_fed_socket("good.domain")
+ FedRegistry.get_fed_socket(@good_domain_origin)
- assert origin == "good.domain"
+ assert origin == "good.domain:80"
Process.exit(pid, :testing)
Process.sleep(100)
- assert {:error, :missing} = FedRegistry.get_fed_socket("good.domain")
+ assert {:error, :missing} = FedRegistry.get_fed_socket(@good_domain_origin)
end
end
- def build_test_socket(origin) do
- uri = FedSockets.uri_for_origin(origin)
+ def build_test_socket(uri) do
Kernel.spawn(fn -> fed_socket_almost(uri) end)
end
def fed_socket_almost(origin) do
FedRegistry.add_fed_socket(origin)
receive do
:close ->
:ok
after
5_000 -> :timeout
end
end
end
diff --git a/test/web/fed_sockets/socket_info_test.exs b/test/web/fed_sockets/socket_info_test.exs
index 3e523c3a6..db3d6edcd 100644
--- a/test/web/fed_sockets/socket_info_test.exs
+++ b/test/web/fed_sockets/socket_info_test.exs
@@ -1,126 +1,118 @@
# 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.SocketInfoTest do
use ExUnit.Case
alias Pleroma.Web.FedSockets
alias Pleroma.Web.FedSockets.SocketInfo
describe "uri_for_origin" do
test "provides the fed_socket URL given the origin information" do
endpoint = "example.com:4000"
assert FedSockets.uri_for_origin(endpoint) =~ "ws://"
assert FedSockets.uri_for_origin(endpoint) =~ endpoint
end
end
describe "origin" do
test "will provide the origin field given a url" do
endpoint = "example.com:4000"
assert SocketInfo.origin("ws://#{endpoint}") == endpoint
assert SocketInfo.origin("http://#{endpoint}") == endpoint
assert SocketInfo.origin("https://#{endpoint}") == endpoint
end
test "will proide the origin field given a uri" do
endpoint = "example.com:4000"
uri = URI.parse("http://#{endpoint}")
assert SocketInfo.origin(uri) == endpoint
end
end
describe "touch" do
test "will update the TTL" do
endpoint = "example.com:4000"
socket = SocketInfo.build("ws://#{endpoint}")
Process.sleep(2)
touched_socket = SocketInfo.touch(socket)
assert socket.connected_until < touched_socket.connected_until
end
end
describe "expired?" do
setup do
start_supervised(
{Pleroma.Web.FedSockets.Supervisor,
[
ping_interval: 8,
connection_duration: 5,
rejection_duration: 5,
fed_socket_rejections: [lazy: true]
]}
)
:ok
end
test "tests if the TTL is exceeded" do
endpoint = "example.com:4000"
socket = SocketInfo.build("ws://#{endpoint}")
refute SocketInfo.expired?(socket)
Process.sleep(10)
assert SocketInfo.expired?(socket)
end
end
describe "creating outgoing connection records" do
test "can be passed a string" do
assert %{conn_pid: :pid, origin: _origin} = SocketInfo.build("example.com:4000", :pid)
end
test "can be passed a URI" do
uri = URI.parse("http://example.com:4000")
assert %{conn_pid: :pid, origin: origin} = SocketInfo.build(uri, :pid)
assert origin =~ "example.com:4000"
end
test "will include the port number" do
assert %{conn_pid: :pid, origin: origin} = SocketInfo.build("http://example.com:4000", :pid)
assert origin =~ ":4000"
end
- test "will not include port 80" do
- assert %{conn_pid: :pid, origin: origin} = SocketInfo.build("http://example.com:80", :pid)
-
- refute origin =~ ":80"
- end
-
- test "does not require the port" do
- assert %{conn_pid: :pid, origin: "example.com"} =
+ test "will provide the port if missing" do
+ assert %{conn_pid: :pid, origin: "example.com:80"} =
SocketInfo.build("http://example.com", :pid)
+
+ assert %{conn_pid: :pid, origin: "example.com:443"} =
+ SocketInfo.build("https://example.com", :pid)
end
end
describe "creating incoming connection records" do
test "can be passed a string" do
assert %{pid: _, origin: _origin} = SocketInfo.build("example.com:4000")
end
test "can be passed a URI" do
uri = URI.parse("example.com:4000")
assert %{pid: _, origin: _origin} = SocketInfo.build(uri)
end
test "will include the port number" do
assert %{pid: _, origin: origin} = SocketInfo.build("http://example.com:4000")
assert origin =~ ":4000"
end
- test "will not include port 80" do
- assert %{pid: _, origin: origin} = SocketInfo.build("http://example.com:80")
-
- refute origin =~ ":80"
- end
-
- test "does not require the port" do
- assert %{pid: _, origin: "example.com"} = SocketInfo.build("http://example.com")
+ test "will provide the port if missing" do
+ assert %{pid: _, origin: "example.com:80"} = SocketInfo.build("http://example.com")
+ assert %{pid: _, origin: "example.com:443"} = SocketInfo.build("https://example.com")
end
end
end

File Metadata

Mime Type
text/x-diff
Expires
Sat, Oct 31, 5:13 AM (18 h, 42 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1801065
Default Alt Text
(20 KB)

Event Timeline