visibility.ex (4822B)
- # Pleroma: A lightweight social networking server
- # Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
- # SPDX-License-Identifier: AGPL-3.0-only
- defmodule Pleroma.Web.ActivityPub.Visibility do
- alias Pleroma.Activity
- alias Pleroma.Object
- alias Pleroma.Repo
- alias Pleroma.User
- alias Pleroma.Web.ActivityPub.Utils
- require Pleroma.Constants
- @spec public?(Object.t() | Activity.t() | map()) :: boolean()
- def public?(%Object{data: %{"type" => "Tombstone"}}), do: false
- def public?(%Object{data: data}), do: public?(data)
- def public?(%Activity{data: %{"type" => "Move"}}), do: true
- def public?(%Activity{data: data}), do: public?(data)
- def public?(%{"directMessage" => true}), do: false
- def public?(data) do
- Utils.label_in_message?(Pleroma.Constants.as_public(), data) or
- Utils.label_in_message?(Utils.as_local_public(), data)
- end
- def local_public?(%Object{data: data}), do: local_public?(data)
- def local_public?(%Activity{data: data}), do: local_public?(data)
- def local_public?(data) do
- Utils.label_in_message?(Utils.as_local_public(), data) and
- not Utils.label_in_message?(Pleroma.Constants.as_public(), data)
- end
- def private?(activity) do
- with false <- public?(activity),
- %User{follower_address: follower_address} <-
- User.get_cached_by_ap_id(activity.data["actor"]) do
- follower_address in activity.data["to"]
- else
- _ -> false
- end
- end
- def announceable?(activity, user, public \\ true) do
- public?(activity) ||
- (!public && private?(activity) && activity.data["actor"] == user.ap_id)
- end
- def direct?(%Activity{data: %{"directMessage" => true}}), do: true
- def direct?(%Object{data: %{"directMessage" => true}}), do: true
- def direct?(activity) do
- !public?(activity) && !private?(activity)
- end
- def list?(%{data: %{"listMessage" => _}}), do: true
- def list?(_), do: false
- @spec visible_for_user?(Object.t() | Activity.t() | nil, User.t() | nil) :: boolean()
- def visible_for_user?(%Object{data: %{"type" => "Tombstone"}}, _), do: false
- def visible_for_user?(%Activity{actor: ap_id}, %User{ap_id: ap_id}), do: true
- def visible_for_user?(%Object{data: %{"actor" => ap_id}}, %User{ap_id: ap_id}), do: true
- def visible_for_user?(nil, _), do: false
- def visible_for_user?(%Activity{data: %{"listMessage" => _}}, nil), do: false
- def visible_for_user?(
- %Activity{data: %{"listMessage" => list_ap_id}} = activity,
- %User{} = user
- ) do
- user.ap_id in activity.data["to"] ||
- list_ap_id
- |> Pleroma.List.get_by_ap_id()
- |> Pleroma.List.member?(user)
- end
- def visible_for_user?(%{__struct__: module} = message, nil)
- when module in [Activity, Object] do
- if restrict_unauthenticated_access?(message),
- do: false,
- else: public?(message) and not local_public?(message)
- end
- def visible_for_user?(%{__struct__: module} = message, user)
- when module in [Activity, Object] do
- x = [user.ap_id | User.following(user)]
- y = [message.data["actor"]] ++ message.data["to"] ++ (message.data["cc"] || [])
- user_is_local = user.local
- federatable = not local_public?(message)
- (public?(message) || Enum.any?(x, &(&1 in y))) and (user_is_local || federatable)
- end
- def entire_thread_visible_for_user?(%Activity{} = activity, %User{} = user) do
- {:ok, %{rows: [[result]]}} =
- Ecto.Adapters.SQL.query(Repo, "SELECT thread_visibility($1, $2)", [
- user.ap_id,
- activity.data["id"]
- ])
- result
- end
- def restrict_unauthenticated_access?(%Activity{local: local}) do
- restrict_unauthenticated_access_to_activity?(local)
- end
- def restrict_unauthenticated_access?(%Object{} = object) do
- object
- |> Object.local?()
- |> restrict_unauthenticated_access_to_activity?()
- end
- def restrict_unauthenticated_access?(%User{} = user) do
- User.visible_for(user, _reading_user = nil)
- end
- defp restrict_unauthenticated_access_to_activity?(local?) when is_boolean(local?) do
- cfg_key = if local?, do: :local, else: :remote
- Pleroma.Config.restrict_unauthenticated_access?(:activities, cfg_key)
- end
- def get_visibility(object) do
- to = object.data["to"] || []
- cc = object.data["cc"] || []
- cond do
- Pleroma.Constants.as_public() in to ->
- "public"
- Pleroma.Constants.as_public() in cc ->
- "unlisted"
- Utils.as_local_public() in to ->
- "local"
- # this should use the sql for the object's activity
- Enum.any?(to, &String.contains?(&1, "/followers")) ->
- "private"
- object.data["directMessage"] == true ->
- "direct"
- is_binary(object.data["listMessage"]) ->
- "list"
- length(cc) > 0 ->
- "private"
- true ->
- "direct"
- end
- end
- end