pleroma/lib/pleroma/object/containment.ex

# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only

defmodule Pleroma.Object.Containment do
  @moduledoc """
  This module contains some useful functions for containing objects to specific
  origins and determining those origins.  They previously lived in the
  ActivityPub `Transmogrifier` module.

  Object containment is an important step in validating remote objects to prevent
  spoofing, therefore removal of object containment functions is NOT recommended.
  """
  def get_actor(%{"actor" => actor}) when is_binary(actor) do
    actor
  end

  def get_actor(%{"actor" => nil, "attributedTo" => actor}) when not is_nil(actor) do
    get_actor(%{"actor" => actor})
  end

  def get_actor(%{"actor" => nil}), do: nil

  def get_actor(%{"actor" => []}), do: nil

  def get_actor(%{"actor" => actor}) when is_list(actor) do
    case actor do
      [first | _] when is_binary(first) ->
        first

      _ ->
        Enum.find_value(actor, fn
          %{"type" => type, "id" => id}
          when type in ["Person", "Service", "Application"] and is_binary(id) ->
            id

          _ ->
            nil
        end)
    end
  end

  def get_actor(%{"actor" => %{"id" => id}}) when is_bitstring(id) do
    id
  end

  # remote data can carry any shape here; callers treat nil as "no actor"
  def get_actor(_), do: nil

  def get_object(%{"object" => id}) when is_binary(id) do
    id
  end

  def get_object(%{"object" => %{"id" => id}}) when is_binary(id) do
    id
  end

  def get_object(_) do
    nil
  end

  defp compare_uris(%URI{host: host} = _id_uri, %URI{host: host} = _other_uri), do: :ok
  defp compare_uris(_id_uri, _other_uri), do: :error

  @doc """
  Checks whether an URL to fetch from is from the local server.

  We never want to fetch from ourselves; if it's not in the database
  it can't be authentic and must be a counterfeit.
  """
  def contain_local_fetch(id) do
    case compare_uris(URI.parse(id), Pleroma.Web.Endpoint.struct_url()) do
      :ok -> :error
      _ -> :ok
    end
  end

  @doc """
  Checks that an imported AP object's actor matches the host it came from.
  """
  def contain_origin(_id, %{"actor" => nil}), do: :error

  def contain_origin(id, %{"actor" => _actor} = params) do
    id_uri = URI.parse(id)

    case get_actor(params) do
      actor when is_binary(actor) -> compare_uris(URI.parse(actor), id_uri)
      _ -> :error
    end
  end

  def contain_origin(id, %{"attributedTo" => actor} = params),
    do: contain_origin(id, Map.put(params, "actor", actor))

  def contain_origin(_id, _data), do: :error

  def contain_origin_from_id(id, %{"id" => other_id} = _params) when is_binary(other_id) do
    id_uri = URI.parse(id)
    other_uri = URI.parse(other_id)

    compare_uris(id_uri, other_uri)
  end

  # Mastodon pin activities don't have an id, so we check the object field, which will be pinned.
  def contain_origin_from_id(id, %{"object" => object}) when is_binary(object) do
    id_uri = URI.parse(id)
    object_uri = URI.parse(object)

    compare_uris(id_uri, object_uri)
  end

  def contain_origin_from_id(_id, _data), do: :error

  @doc """
  Checks that a fetched document's id is the URL it was fetched from
  (after redirects).

  Fragments, a trailing slash, the case of the host and differences
  that our URL encoding removes before a request are ignored.
  Documents without an id never match.

  Based on Akkoma's `contain_id_to_fetch/2`.
  """
  def contain_id_to_fetch(url, %{"id" => id}) when is_binary(url) and is_binary(id) do
    if normalize_uri(url) == normalize_uri(id), do: :ok, else: :error
  end

  def contain_id_to_fetch(_url, _data), do: :error

  defp normalize_uri(uri) do
    uri = uri |> Pleroma.Utils.URIEncoding.encode_url() |> URI.parse()

    %URI{
      uri
      | fragment: nil,
        host: uri.host && String.downcase(uri.host, :ascii),
        path: uri.path && String.replace_suffix(uri.path, "/", "")
    }
    |> URI.to_string()
  end

  def contain_child(%{"object" => %{"id" => id, "attributedTo" => _} = object}),
    do: contain_origin(id, object)

  def contain_child(_), do: :ok

  @doc "Checks that two URIs are on the same host. URIs without a host never match."
  def same_origin(id, other_id) when is_binary(id) and is_binary(other_id) do
    case {URI.parse(id), URI.parse(other_id)} do
      {%URI{host: host}, %URI{host: host}} when is_binary(host) and host != "" -> :ok
      _ -> :error
    end
  end

  def same_origin(_id, _other_id), do: :error
end