media_proxy.ex
View Options

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

	defmodule Pleroma.Web.MediaProxy do
	alias Pleroma.Config
	alias Pleroma.Helpers.UriHelper
	alias Pleroma.Upload
	alias Pleroma.Web.Endpoint
	alias Pleroma.Web.MediaProxy.Invalidation

	@base64_opts [padding: false]
	@cache_table :banned_urls_cache

	@cachex Pleroma.Config.get([:cachex, :provider], Cachex)

	def cache_table, do: @cache_table

	@spec in_banned_urls(String.t()) :: boolean()
	def in_banned_urls(url), do: elem(@cachex.exists?(@cache_table, url(url)), 1)

	def remove_from_banned_urls(urls) when is_list(urls) do
	@cachex.execute!(@cache_table, fn cache ->
	Enum.each(Invalidation.prepare_urls(urls), &@cachex.del(cache, &1))
	end)
	end

	def remove_from_banned_urls(url) when is_binary(url) do
	@cachex.del(@cache_table, url(url))
	end

	def put_in_banned_urls(urls) when is_list(urls) do
	@cachex.execute!(@cache_table, fn cache ->
	Enum.each(Invalidation.prepare_urls(urls), &@cachex.put(cache, &1, true))
	end)
	end

	def put_in_banned_urls(url) when is_binary(url) do
	@cachex.put(@cache_table, url(url), true)
	end

	def url(url) when is_nil(url) or url == "", do: nil
	def url("/" <> _ = url), do: url

	def url(url) do
	if enabled?() and url_proxiable?(url) do
	encode_url(url)
	else
	url
	end
	end

	@spec url_proxiable?(String.t()) :: boolean()
	def url_proxiable?(url) do
	not local?(url) and not whitelisted?(url)
	end

	def preview_url(url, preview_params \\ []) do
	if preview_enabled?() do
	encode_preview_url(url, preview_params)
	else
	url(url)
	end
	end

	def enabled?, do: Config.get([:media_proxy, :enabled], false)

	# Note: media proxy must be enabled for media preview proxy in order to load all
	# non-local non-whitelisted URLs through it and be sure that body size constraint is preserved.
	def preview_enabled?, do: enabled?() and !!Config.get([:media_preview_proxy, :enabled])

	def local?(url), do: String.starts_with?(url, Endpoint.url())

	def whitelisted?(url) do
	%{host: domain} = URI.parse(url)

	mediaproxy_whitelist_domains =
	Config.get([:media_proxy, :whitelist], [])
	\|> Kernel.++(["#{Upload.base_url()}"])
	\|> Enum.map(&maybe_get_domain_from_url/1)

	domain in mediaproxy_whitelist_domains
	end

	defp maybe_get_domain_from_url("http" <> _ = url) do
	URI.parse(url).host
	end

	defp maybe_get_domain_from_url(domain), do: domain

	defp base64_sig64(url) do
	base64 = Base.url_encode64(url, @base64_opts)

	sig64 =
	base64
	\|> signed_url()
	\|> Base.url_encode64(@base64_opts)

	{base64, sig64}
	end

	def encode_url(url) do
	{base64, sig64} = base64_sig64(url)

	build_url(sig64, base64, filename(url))
	end

	def encode_preview_url(url, preview_params \\ []) do
	{base64, sig64} = base64_sig64(url)

	build_preview_url(sig64, base64, filename(url), preview_params)
	end

	def decode_url(sig, url) do
	with {:ok, sig} <- Base.url_decode64(sig, @base64_opts),
	signature when signature == sig <- signed_url(url) do
	{:ok, Base.url_decode64!(url, @base64_opts)}
	else
	_ -> {:error, :invalid_signature}
	end
	end

	def decode_url(encoded) do
	[_, "proxy", sig, base64 \| _] = URI.parse(encoded).path \|> String.split("/")
	decode_url(sig, base64)
	end

	defp signed_url(url) do
	:crypto.mac(:hmac, :sha, Config.get([Endpoint, :secret_key_base]), url)
	end

	def filename(url_or_path) do
	if path = URI.parse(url_or_path).path, do: Path.basename(path)
	end

	def base_url do
	Config.get([:media_proxy, :base_url], Endpoint.url())
	end

	defp proxy_url(path, sig_base64, url_base64, filename) do
	[
	base_url(),
	path,
	sig_base64,
	url_base64,
	filename
	]
	\|> Enum.filter(& &1)
	\|> Path.join()
	end

	def build_url(sig_base64, url_base64, filename \\ nil) do
	proxy_url("proxy", sig_base64, url_base64, filename)
	end

	def build_preview_url(sig_base64, url_base64, filename \\ nil, preview_params \\ []) do
	uri = proxy_url("proxy/preview", sig_base64, url_base64, filename)

	UriHelper.modify_uri_params(uri, preview_params)
	end

	def verify_request_path_and_url(
	%Plug.Conn{params: %{"filename" => _}, request_path: request_path},
	url
	) do
	verify_request_path_and_url(request_path, url)
	end

	def verify_request_path_and_url(request_path, url) when is_binary(request_path) do
	filename = filename(url)

	if filename && not basename_matches?(request_path, filename) do
	{:wrong_filename, filename}
	else
	:ok
	end
	end

	def verify_request_path_and_url(_, _), do: :ok

	defp basename_matches?(path, filename) do
	basename = Path.basename(path)
	basename == filename or URI.decode(basename) == filename or URI.encode(basename) == filename
	end
	end

File Metadata

Mime Type: text/x-ruby
Expires: Fri, Feb 21, 5:42 PM (13 h, 12 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 66793
Default Alt Text: media_proxy.ex (4 KB)

media_proxy.ex
No OneTemporary
Actions

media_proxy.ex
View Options

File Metadata

Event Timeline

media_proxy.exNo OneTemporaryActions

media_proxy.exView Options

File Metadata

Event Timeline

media_proxy.ex
No OneTemporary
Actions

media_proxy.ex
View Options