Merge branch 'develop' into mastodon-admin-api

This commit is contained in:
lain 2026-07-26 07:46:55 +00:00
commit 51349f61f8
216 changed files with 9028 additions and 921 deletions

View file

@ -1,5 +1,6 @@
when:
- event: pull_request
evaluate: 'CI_COMMIT_SOURCE_BRANCH != "weblate"'
labels:
platform: linux/amd64

View file

@ -27,7 +27,7 @@ steps:
settings: &docker_settings
registry: "git.pleroma.social"
image: "pleroma/pleroma"
architectures: [amd64, arm64]
architectures: [amd64, arm64, armv7]
tags:
- latest
- develop

View file

@ -12,6 +12,7 @@ matrix:
platform:
- linux/amd64
- linux/arm64
- linux/arm
# This is needed for the when clauses below.
labels:
@ -26,11 +27,19 @@ variables:
from_secret: pleroma-ci-user
password:
from_secret: pleroma-ci-password
kaniko_image: &kaniko_image woodpeckerci/plugin-kaniko:2.3.1
armv7_docker_variables: &armv7_docker_variables
<<: *docker_variables
build_args:
- ELIXIR_IMG=docker.io/arm32v7/elixir
- ELIXIR_TAG=1.17.3-otp-26-alpine
- ALPINE_IMG=docker.io/arm32v7/alpine
- ALPINE_VER=3.20
extra_opts: --custom-platform=linux/arm/v7
kaniko_image: &kaniko_image docker.io/woodpeckerci/plugin-kaniko:2.3.2
steps:
docker-develop-amd64:
image: woodpeckerci/plugin-kaniko:2.3.1
image: *kaniko_image
when:
- evaluate: 'platform == "linux/amd64" && CI_COMMIT_BRANCH == "${CI_REPO_DEFAULT_BRANCH}"'
settings:
@ -41,7 +50,7 @@ steps:
- ${CI_COMMIT_SHA:0:8}-amd64
docker-develop-arm64:
image: woodpeckerci/plugin-kaniko:2.3.1
image: *kaniko_image
when:
- evaluate: 'platform == "linux/arm64" && CI_COMMIT_BRANCH == "${CI_REPO_DEFAULT_BRANCH}"'
settings:
@ -51,6 +60,17 @@ steps:
- develop-arm64
- ${CI_COMMIT_SHA:0:8}-arm64
docker-develop-armv7:
image: *kaniko_image
when:
- evaluate: 'platform == "linux/arm" && CI_COMMIT_BRANCH == "${CI_REPO_DEFAULT_BRANCH}"'
settings:
<<: *armv7_docker_variables
tags:
- latest-armv7
- develop-armv7
- ${CI_COMMIT_SHA:0:8}-armv7
docker-stable-amd64:
image: *kaniko_image
when:
@ -62,6 +82,18 @@ steps:
- stable-amd64
- ${CI_COMMIT_SHA:0:8}-amd64
# FE workflow runs only on linux/amd64
docker-stable-e2e-amd64:
image: *kaniko_image
when:
- evaluate: 'platform == "linux/amd64" && CI_PIPELINE_EVENT == "tag"'
- evaluate: 'platform == "linux/amd64" && CI_PIPELINE_EVENT == "manual" && CI_COMMIT_BRANCH == "stable"'
settings:
<<: *docker_variables
dockerfile: Dockerfile-e2e
tags:
- stable-e2e
docker-stable-tag-amd64:
image: *kaniko_image
when:
@ -94,3 +126,25 @@ steps:
tags:
- <<: *arm64_tags
- ${CI_COMMIT_TAG}-arm64
docker-stable-armv7:
image: *kaniko_image
when:
- evaluate: 'platform == "linux/arm" && CI_PIPELINE_EVENT == "manual" && CI_COMMIT_BRANCH == "stable" && CI_COMMIT_TAG == ""'
settings:
<<: *armv7_docker_variables
tags: &armv7_tags
- latest-armv7
- stable-armv7
- ${CI_COMMIT_SHA:0:8}-armv7
docker-stable-tag-armv7:
image: *kaniko_image
when:
- evaluate: 'platform == "linux/arm" && CI_PIPELINE_EVENT == "tag"'
- evaluate: 'platform == "linux/arm" && CI_PIPELINE_EVENT == "manual" && CI_COMMIT_BRANCH == "stable" && CI_COMMIT_TAG != ""'
settings:
<<: *armv7_docker_variables
tags:
- <<: *armv7_tags
- ${CI_COMMIT_TAG}-armv7

View file

@ -1,6 +1,7 @@
when:
- event: pull_request
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]
evaluate: 'CI_COMMIT_SOURCE_BRANCH != "weblate"'
- event: push
branch: ${CI_REPO_DEFAULT_BRANCH}
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]

View file

@ -1,6 +1,7 @@
when:
- event: pull_request
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]
evaluate: 'CI_COMMIT_SOURCE_BRANCH != "weblate"'
- event: push
branch: ${CI_REPO_DEFAULT_BRANCH}
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]
@ -23,10 +24,11 @@ steps:
entrypoint: *script_file_entrypoint
environment:
MIX_ENV: test
VIX_COMPILATION_MODE: PLATFORM_PROVIDED_LIBVIPS
DB_HOST: postgres
DB_PORT: 5432
commands:
- apk add --no-cache build-base cmake exiftool ffmpeg file-dev git openssl
- apk add --no-cache build-base cmake exiftool ffmpeg vips-dev file-dev git openssl
- adduser -D -h /home/testuser testuser
- mkdir -p /home/testuser/.mix /home/testuser/.hex
- chown -R testuser:testuser . /home/testuser

View file

@ -1,6 +1,7 @@
when:
- event: pull_request
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]
evaluate: 'CI_COMMIT_SOURCE_BRANCH != "weblate"'
- event: push
branch: ${CI_REPO_DEFAULT_BRANCH}
path: [ "**/*.ex", "**/*.eex", "**/*.exs", "mix.lock", ".woodpecker/**" ]
@ -18,15 +19,16 @@ variables:
- 'printf "%s" "$CI_SCRIPT" | base64 -d > /tmp/ci-script.sh && /bin/sh -xe /tmp/ci-script.sh'
steps:
unit-testing-elixir-1.18:
image: elixir:1.18-otp-27-alpine
unit-testing-elixir-1.19:
image: elixir:1.19-otp-28-alpine
entrypoint: *script_file_entrypoint
environment:
MIX_ENV: test
VIX_COMPILATION_MODE: PLATFORM_PROVIDED_LIBVIPS
DB_HOST: postgres
DB_PORT: 5432
commands:
- apk add --no-cache build-base cmake exiftool ffmpeg file-dev git openssl
- apk add --no-cache build-base cmake exiftool ffmpeg vips-dev file-dev git openssl
- adduser -D -h /home/testuser testuser
- mkdir -p /home/testuser/.mix /home/testuser/.hex
- chown -R testuser:testuser . /home/testuser

View file

@ -1,10 +1,12 @@
# https://hub.docker.com/r/hexpm/elixir/tags
ARG ELIXIR_IMG=hexpm/elixir
ARG ELIXIR_VER=1.17.3
ARG ERLANG_VER=26.2.5.6
ARG ALPINE_VER=3.17.9
ARG ERLANG_VER=27.3.4.14
ARG ALPINE_VER=3.23.5
ARG ELIXIR_TAG=${ELIXIR_VER}-erlang-${ERLANG_VER}-alpine-${ALPINE_VER}
ARG ALPINE_IMG=alpine
FROM ${ELIXIR_IMG}:${ELIXIR_VER}-erlang-${ERLANG_VER}-alpine-${ALPINE_VER} AS build
FROM ${ELIXIR_IMG}:${ELIXIR_TAG} AS build
COPY . .
@ -20,7 +22,7 @@ RUN apk add git gcc g++ musl-dev make cmake file-dev vips-dev &&\
mkdir release &&\
mix release --path release
FROM alpine:${ALPINE_VER}
FROM ${ALPINE_IMG}:${ALPINE_VER}
ARG BUILD_DATE
ARG VCS_REF

61
Dockerfile-e2e Normal file
View file

@ -0,0 +1,61 @@
# https://hub.docker.com/r/hexpm/elixir/tags
ARG ELIXIR_IMG=docker.io/hexpm/elixir
ARG ALPINE_IMG=docker.io/alpine
ARG ELIXIR_VER=1.18.4
ARG ERLANG_VER=27.3.4.11
ARG ALPINE_VER=3.23.4
FROM ${ELIXIR_IMG}:${ELIXIR_VER}-erlang-${ERLANG_VER}-alpine-${ALPINE_VER} AS build
COPY . .
ENV MIX_ENV=prod
ENV VIX_COMPILATION_MODE=PLATFORM_PROVIDED_LIBVIPS
RUN apk add git gcc g++ musl-dev make cmake file-dev vips-dev &&\
echo "import Config" > config/prod.secret.exs &&\
mix local.hex --force &&\
mix local.rebar --force &&\
mix deps.clean --all &&\
mix deps.get --only prod &&\
mkdir release &&\
mix release --path release
FROM ${ALPINE_IMG}:${ALPINE_VER}
ARG BUILD_DATE
ARG VCS_REF
LABEL maintainer="ops@pleroma.social" \
org.opencontainers.image.title="pleroma" \
org.opencontainers.image.description="Pleroma FE E2E test image" \
org.opencontainers.image.authors="ops@pleroma.social" \
org.opencontainers.image.vendor="pleroma.social" \
org.opencontainers.image.documentation="https://git.pleroma.social/pleroma/pleroma" \
org.opencontainers.image.licenses="AGPL-3.0" \
org.opencontainers.image.url="https://pleroma.social" \
org.opencontainers.image.revision=$VCS_REF \
org.opencontainers.image.created=$BUILD_DATE
ARG HOME=/opt/pleroma
ARG DATA=/var/lib/pleroma
RUN apk update &&\
apk add exiftool ffmpeg vips libmagic ncurses postgresql-client &&\
adduser --system --shell /bin/false --home ${HOME} pleroma &&\
mkdir -p ${DATA}/uploads &&\
mkdir -p ${DATA}/static &&\
chown -R pleroma ${DATA} &&\
mkdir -p /etc/pleroma &&\
chown -R pleroma /etc/pleroma
USER pleroma
COPY --from=build --chown=pleroma:0 /release ${HOME}
COPY --chown=pleroma --chmod=640 ./config/docker.exs /etc/pleroma/config.exs
COPY ./docker-entrypoint-e2e.sh ${HOME}
EXPOSE 4000
ENTRYPOINT ["/opt/pleroma/docker-entrypoint-e2e.sh"]

View file

@ -0,0 +1 @@
Fixed `/api/v1/pleroma/accounts/:id/favourites` ordering and pagination to use the favourite date.

View file

@ -0,0 +1 @@
Switch native captcha to the published pleroma_captcha Hex package.

View file

@ -0,0 +1 @@
Allow disabling Chats feature

View file

@ -0,0 +1 @@
Dockerfile: update base images to fix problems with SSL handshakes

View file

View file

@ -0,0 +1 @@
Implemented ConfigDB configuration blacklist and accompanying Mix task

View file

@ -0,0 +1 @@
Fixed Elixir 1.19 warnings

View file

@ -0,0 +1 @@
Support fetching featured collection items from the first collection page

View file

@ -0,0 +1 @@
Fix preferred frontend not working due to InstanceStatic plug prematurely returning HTTP 404

View file

@ -0,0 +1 @@
Add Mastodon-compatible grouped notifications API endpoints under `/api/v2/notifications` for newly-created notifications

View file

@ -0,0 +1 @@
Fixed Gun HTTP/1 concurrency, forwarding and authenticated proxy handling, tunnel requests, and streamed connection cleanup. Updated Gun and Cowlib for HTTP client security fixes.

View file

@ -0,0 +1 @@
Fix hashtag timeline query for additional tag handling

View file

@ -0,0 +1 @@
Ensure Host header is present and matches instance URI

View file

@ -0,0 +1 @@
Handle reports with just actor ap id as the object

View file

@ -0,0 +1 @@
Allow attaching emojis to user lists

View file

@ -0,0 +1 @@
Changed how Logger is configured due to changes in Elixir and added deprecation warnings for the old configuration format

View file

@ -0,0 +1 @@
Update majic to 1.2.0.

View file

@ -0,0 +1 @@
Echo Mastodon-style `Sec-WebSocket-Protocol` tokens in streaming WebSocket handshakes.

View file

@ -0,0 +1 @@
Detect image content types for extensionless MediaProxy responses.

View file

View file

@ -0,0 +1 @@
Avoid a duplicate user query during OAuth authentication.

View file

@ -0,0 +1 @@
Use the Hex package for oban_plugins_lazarus.

View file

@ -0,0 +1 @@
Switch patched Phoenix 1.7.14 back to upstream with Phoenix 1.8.0+

View file

@ -0,0 +1 @@
Avoid creating atoms from serialized PublisherWorker parameters.

View file

@ -0,0 +1 @@
Reject incoming reports when both the reporter and reported account are remote

View file

@ -0,0 +1 @@
Use upstream Hex releases for the `remote_ip` dependency and expose client IP ranges for remote IP resolution.

1
changelog.d/rss.fix Normal file
View file

@ -0,0 +1 @@
Fix RSS feeds to be W3C compliant and render better in RSS readers

View file

@ -0,0 +1 @@
Keep followed users visible in streams when their domain is blocked.

View file

View file

@ -0,0 +1 @@
Fix inconsistent User cache keys and invalidate numeric ID aliases.

View file

@ -0,0 +1 @@
Improve user search / autocompletion ordering.

View file

View file

@ -0,0 +1 @@
Include arm32v7 in multi-arch Docker images

View file

@ -0,0 +1 @@
RichMedia: Fix backfill causing old posts to show up on timelines by disabling it in MastoAPI StatusView

View file

@ -133,13 +133,17 @@ config :pleroma, Pleroma.Web.Endpoint,
]
# Configures Elixir's Logger
config :logger, backends: [:console]
# Primary config
config :logger, level: :debug
config :logger, :console,
level: :debug,
# Console config
config :logger, :default_handler, level: :debug
config :logger, :default_formatter,
format: "\n$time $metadata[$level] $message\n",
metadata: [:actor, :path, :type, :user]
# Syslog config
config :logger, :ex_syslogger,
level: :debug,
ident: "pleroma",
@ -738,6 +742,7 @@ config :pleroma, Pleroma.Workers.PurgeExpiredActivity, enabled: true, min_lifeti
config :pleroma, Pleroma.Web.Plugs.RemoteIp,
enabled: true,
headers: ["x-forwarded-for"],
clients: [],
proxies: [],
reserved: [
"127.0.0.0/8",
@ -961,6 +966,10 @@ config :pleroma, Pleroma.Search.QdrantSearch,
vectors: %{size: 384, distance: "Cosine"}
}
config :pleroma, :database_config_blacklist, [
{:pleroma, :logger}
]
config :pleroma, :database_config_whitelist, [
{:pleroma},
{:cors_plug},
@ -970,6 +979,8 @@ config :pleroma, :database_config_whitelist, [
{:web_push_encryption, :vapid_details}
]
config :pleroma, Pleroma.Chat, enabled: true
# Import environment specific config. This must remain at the bottom
# of this file so it overrides the configuration defined above.
import_config "#{Mix.env()}.exs"

View file

@ -2726,6 +2726,19 @@ config :pleroma, :config_description, [
}
]
},
%{
group: :pleroma,
key: Pleroma.Chat,
type: :group,
description: "Pleroma Chat settings",
children: [
%{
key: :enabled,
type: :boolean,
description: "Enables the chats API."
}
]
},
%{
group: :pleroma,
key: :http,
@ -2910,7 +2923,7 @@ config :pleroma, :config_description, [
key: Pleroma.Web.Plugs.RemoteIp,
type: :group,
description: """
`Pleroma.Web.Plugs.RemoteIp` is a shim to call [`RemoteIp`](https://git.pleroma.social/pleroma/remote_ip) but with runtime configuration.
`Pleroma.Web.Plugs.RemoteIp` is a shim to call [`RemoteIp`](https://hex.pm/packages/remote_ip) but with runtime configuration.
**If your instance is not behind at least one reverse proxy, you should not enable this plug.**
""",
children: [
@ -2926,6 +2939,12 @@ config :pleroma, :config_description, [
A list of strings naming the HTTP headers to use when deriving the true client IP. Default: `["x-forwarded-for"]`.
"""
},
%{
key: :clients,
type: {:list, :string},
description:
"A list of client IPs or subnets in CIDR notation. These will not be treated as proxies or reserved ranges. Defaults to `[]`. IPv4 entries without a bitmask will be assumed to be /32 and IPv6 /128."
},
%{
key: :proxies,
type: {:list, :string},
@ -3079,9 +3098,16 @@ config :pleroma, :config_description, [
%{
key: :max_connections,
type: :integer,
description: "Maximum number of connections in the pool. Default: 250 connections.",
description:
"Maximum total number of connections in the pool. HTTP/1 origins may use multiple connections within this limit, while HTTP/2 connections are multiplexed. Default: 250 connections.",
suggestions: [250]
},
%{
key: :max_idle_time,
type: :integer,
description: "Time before an unused connection is closed. Default: 30000ms.",
suggestions: [30_000]
},
%{
key: :connect_timeout,
type: :integer,

View file

@ -36,7 +36,7 @@ config :pleroma, Pleroma.Emails.Mailer, adapter: Swoosh.Adapters.Local
# different ports.
# Do not include timestamps in development logs
config :logger, Logger.Backends.Console, format: "$metadata[$level] $message\n"
config :logger, :default_formatter, format: "$metadata[$level] $message\n"
# Set a higher stacktrace during development. Avoid configuring such
# in production as building large stacktraces may be expensive.

View file

@ -20,8 +20,7 @@ config :pleroma, Pleroma.Web.Endpoint,
config :phoenix, serve_endpoints: true
# Do not print debug messages in production
config :logger, Logger.Backends.Console, level: :info
config :logger, :console, level: :info
config :logger, :default_handler, level: :info
config :logger, :ex_syslogger, level: :info
# ## SSL Support

View file

@ -15,9 +15,11 @@ config :pleroma, Pleroma.Captcha,
method: Pleroma.Captcha.Mock
# Print only warnings and errors during test
config :logger, :console,
level: :warning,
format: "\n[$level] $message\n"
config :logger, :default_handler, level: :warning
config :logger, :default_formatter,
format: "\n[$level] $message\n",
colors: [enabled: false]
config :pleroma, :auth, oauth_consumer_strategies: []

70
docker-entrypoint-e2e.sh Executable file
View file

@ -0,0 +1,70 @@
#!/bin/ash
set -eu
SEED_SENTINEL_PATH=/var/lib/pleroma/.e2e_seeded
CONFIG_OVERRIDE_PATH=/var/lib/pleroma/config.exs
echo '-- Waiting for database...'
while ! pg_isready -U "${DB_USER:-pleroma}" -d "postgres://${DB_HOST:-db}:${DB_PORT:-5432}/${DB_NAME:-pleroma}" -t 1; do
sleep 1s
done
echo '-- Writing E2E config overrides...'
cat > "$CONFIG_OVERRIDE_PATH" <<EOF
import Config
config :pleroma, Pleroma.Captcha,
enabled: false
config :pleroma, :instance,
registrations_open: true,
account_activation_required: false,
account_approval_required: false
EOF
echo '-- Running migrations...'
/opt/pleroma/bin/pleroma_ctl migrate
if [ ! -f "$SEED_SENTINEL_PATH" ]; then
if [ -n "${E2E_ADMIN_USERNAME:-}" ] && [ -n "${E2E_ADMIN_PASSWORD:-}" ] && [ -n "${E2E_ADMIN_EMAIL:-}" ]; then
echo '-- Seeding admin user' "$E2E_ADMIN_USERNAME" '...'
if ! PLEROMA_CTL_RPC_DISABLED=true /opt/pleroma/bin/pleroma_ctl user new "$E2E_ADMIN_USERNAME" "$E2E_ADMIN_EMAIL" --admin --password "$E2E_ADMIN_PASSWORD" -y; then
echo '-- User already exists or creation failed, ensuring admin + confirmed...'
PLEROMA_CTL_RPC_DISABLED=true /opt/pleroma/bin/pleroma_ctl user set "$E2E_ADMIN_USERNAME" --admin --confirmed
fi
else
echo '-- Skipping admin seeding (missing E2E_ADMIN_* env)'
fi
touch "$SEED_SENTINEL_PATH"
fi
echo '-- Starting!'
/opt/pleroma/bin/pleroma start &
PLEROMA_PID=$!
cleanup() {
if kill -0 "$PLEROMA_PID" 2>/dev/null; then
kill -TERM "$PLEROMA_PID"
wait "$PLEROMA_PID" || true
fi
}
trap cleanup INT TERM
echo '-- Waiting for API...'
api_ok=false
for _i in $(seq 1 120); do
if wget -qO- http://127.0.0.1:4000/api/v1/instance >/dev/null 2>&1; then
api_ok=true
break
fi
sleep 1s
done
if [ "$api_ok" != true ]; then
echo 'Timed out waiting for Pleroma API to become available'
exit 1
fi
wait "$PLEROMA_PID"

View file

@ -573,7 +573,8 @@ Settings for HTTP connection pool.
* `:connection_acquisition_wait` - Timeout to acquire a connection from pool.The total max time is this value multiplied by the number of retries.
* `connection_acquisition_retries` - Number of attempts to acquire the connection from the pool if it is overloaded. Each attempt is timed `:connection_acquisition_wait` apart.
* `:max_connections` - Maximum number of connections in the pool.
* `:max_connections` - Maximum total number of connections in the pool. HTTP/1 origins may use multiple connections within this limit, while HTTP/2 connections are multiplexed.
* `:max_idle_time` - Time before an unused connection is closed.
* `:connect_timeout` - Timeout to connect to the host.
* `:reclaim_multiplier` - Multiplied by `:max_connections` this will be the maximum number of idle connections that will be reclaimed in case the pool is overloaded.
@ -858,21 +859,23 @@ Web Push Notifications configuration. You can use the mix task `mix web_push.gen
* ``private_key``: VAPID private key
## :logger
* `backends`: `:console` is used to send logs to stdout, `{ExSyslogger, :ex_syslogger}` to log to syslog
* Logging to console/stdout is done by default, use `{ExSyslogger, :ex_syslogger}` to log to syslog
An example to enable ONLY ExSyslogger (f/ex in ``prod.secret.exs``) with info and debug suppressed:
```elixir
config :logger,
config :pleroma, :logger,
backends: [{ExSyslogger, :ex_syslogger}]
config :logger, default_handler: false
config :logger, :ex_syslogger,
level: :warning
```
Another example, keeping console output and adding the pid to syslog output:
```elixir
config :logger,
backends: [:console, {ExSyslogger, :ex_syslogger}]
config :pleroma, :logger,
backends: [{ExSyslogger, :ex_syslogger}]
config :logger, :ex_syslogger,
level: :warning,
@ -883,23 +886,22 @@ See: [logger’s documentation](https://hexdocs.pm/logger/Logger.html) and [ex_s
An example of logging info to local syslog, but debug to console:
```elixir
config :logger,
backends: [ {ExSyslogger, :ex_syslogger}, :console ],
level: :info
config :pleroma, :logger,
backends: [{ExSyslogger, :ex_syslogger}]
config :logger, :ex_syslogger,
level: :info,
ident: "pleroma",
format: "$metadata[$level] $message"
config :logger, :console,
level: :debug,
config :logger, :default_handler,
level: :debug
config :logger, :default_formatter,
format: "\n$time $metadata[$level] $message\n",
metadata: [:request_id]
```
## Database options
### RUM indexing for full text search

View file

@ -10,3 +10,6 @@ The other way to do it, for example, with Tor you would most likely add somethin
```
config :pleroma, :http, proxy_url: {:socks5, :localhost, 9050}
```
The Gun HTTP adapter supports SOCKS5. SOCKS4 proxies require another HTTP adapter.
For regular HTTP proxies, Gun forwards plain HTTP requests directly and uses CONNECT tunnels for HTTPS requests.

View file

@ -1,8 +1,8 @@
## Required dependencies
* PostgreSQL >=11.0
* Elixir >=1.15.0 <1.19
* Erlang OTP >=23.0.0 (supported: <28)
* Elixir >=1.15.0, <1.20
* Erlang OTP >=24.0.0 (supported: <29)
* git
* file / libmagic
* gcc or clang

View file

@ -252,40 +252,43 @@ defmodule Mix.Tasks.Pleroma.Config do
if whitelisted_configs in [nil, false] do
shell_error("No unwanted settings in ConfigDB. No changes made.")
else
whitelisted_groups =
whitelisted_configs
|> Enum.filter(fn
{_group} -> true
_ -> false
end)
|> Enum.map(fn {group} -> group end)
whitelisted_keys =
whitelisted_configs
|> Enum.filter(fn
{_group, _key} -> true
_ -> false
end)
{whitelisted_groups, whitelisted_keys} = get_filters(whitelisted_configs)
filtered =
from(c in ConfigDB)
|> Repo.all()
|> Enum.filter(&not_whitelisted?(&1, whitelisted_groups, whitelisted_keys))
if not Enum.empty?(filtered) do
shell_info("The following settings will be removed from ConfigDB:\n")
Enum.each(filtered, &dump(&1))
do_config_filter(filtered, force)
end
end
if force or shell_prompt("Are you sure you want to continue?", "n") in ~w(Yn Y y) do
filtered_ids = Enum.map(filtered, fn %{id: id} -> id end)
# Removes blacklisted configuration sections
def run(["filter_blacklisted" | rest]) do
{options, [], []} =
OptionParser.parse(
rest,
strict: [force: :boolean],
aliases: [f: :force]
)
Repo.delete_all(from(c in ConfigDB, where: c.id in ^filtered_ids))
else
shell_error("No changes made.")
end
else
shell_error("No unwanted settings in ConfigDB. No changes made.")
end
force = Keyword.get(options, :force, false)
start_pleroma()
blacklisted_configs = Pleroma.Config.get(:database_config_blacklist)
if blacklisted_configs in [nil, false] do
shell_error("No unwanted settings in ConfigDB. No changes made.")
else
{blacklisted_groups, blacklisted_keys} = get_filters(blacklisted_configs)
filtered =
from(c in ConfigDB)
|> Repo.all()
|> Enum.filter(&blacklisted?(&1, blacklisted_groups, blacklisted_keys))
do_config_filter(filtered, force)
end
end
@ -490,8 +493,49 @@ defmodule Mix.Tasks.Pleroma.Config do
Ecto.Adapters.SQL.query!(Repo, "ALTER SEQUENCE config_id_seq RESTART;")
end
defp get_filters(config) do
groups =
config
|> Enum.filter(fn
{_group} -> true
_ -> false
end)
|> Enum.map(fn {group} -> group end)
keys =
config
|> Enum.filter(fn
{_group, _key} -> true
_ -> false
end)
{groups, keys}
end
defp not_whitelisted?(%{group: group, key: key}, whitelisted_groups, whitelisted_keys) do
not Enum.member?(whitelisted_groups, group) and
not Enum.member?(whitelisted_keys, {group, key})
end
defp blacklisted?(%{group: group, key: key}, blacklisted_groups, blacklisted_keys) do
Enum.member?(blacklisted_groups, group) or
Enum.member?(blacklisted_keys, {group, key})
end
defp do_config_filter(filtered, _) when length(filtered) == 0 do
shell_error("No unwanted settings in ConfigDB. No changes made.")
end
defp do_config_filter(filtered, force) do
shell_info("The following settings will be removed from ConfigDB:\n")
Enum.each(filtered, &dump(&1))
if force or shell_prompt("Are you sure you want to continue?", "n") in ~w(Yn Y y) do
filtered_ids = Enum.map(filtered, fn %{id: id} -> id end)
Repo.delete_all(from(c in ConfigDB, where: c.id in ^filtered_ids))
else
shell_error("No changes made.")
end
end
end

View file

@ -43,6 +43,7 @@ defmodule Pleroma.Application do
# every time the application is restarted, so we disable module
# conflicts at runtime
Code.compiler_options(ignore_module_conflict: true)
configure_logger()
Pleroma.Telemetry.Logger.attach()
Config.Holder.save_default()
Pleroma.HTML.compile_scrubbers()
@ -113,6 +114,44 @@ defmodule Pleroma.Application do
Supervisor.start_link(children, opts)
end
def configure_logger do
Config.get([:logger, :backends], [])
|> Enum.each(fn backend ->
case backend_to_logger(backend) do
{:ok, logger} ->
add_logger(logger)
{:error, :console_backend} ->
Logger.warning(":console is no longer considered a backend and is enabled by default")
end
end)
end
defp add_logger(backend) do
case LoggerBackends.add(backend) do
{:ok, _} ->
Logger.debug("Successfully added logger backend: #{inspect(backend)}")
{:error, :already_present} ->
:ok
{:error, reason} ->
Logger.error("Failed to add logger backend #{inspect(backend)}: #{inspect(reason)}")
end
end
defp backend_to_logger(:console), do: {:error, :console_backend}
defp backend_to_logger({:ex_syslogger = backend, name}) do
Logger.warning(
"Configuration {:#{backend}, :#{name}} is incorrect. Use {ExSyslogger, :#{name}} instead!"
)
{:ok, {ExSyslogger, name}}
end
defp backend_to_logger(backend), do: {:ok, backend}
def load_custom_modules do
dir = Config.get([:modules, :runtime_dir])

View file

@ -108,4 +108,6 @@ defmodule Pleroma.Chat do
|> cast(%{pinned: false}, [:pinned])
|> Repo.update()
end
def enabled?, do: Pleroma.Config.get([__MODULE__, :enabled], true)
end

View file

@ -20,6 +20,8 @@ defmodule Pleroma.Config.DeprecationWarnings do
"\n* `config :pleroma, :instance, mrf_transparency_exclusions` is now `config :pleroma, :mrf, transparency_exclusions`"}
]
@logger_formatter_config_knobs [:colors, :format, :metadata, :truncate, :utc_log]
def check_exiftool_filter do
filters = Config.get([Pleroma.Upload]) |> Keyword.get(:filters, [])
@ -218,7 +220,8 @@ defmodule Pleroma.Config.DeprecationWarnings do
check_quarantined_instances_tuples(),
check_transparency_exclusions_tuples(),
check_simple_policy_tuples(),
check_exiftool_filter()
check_exiftool_filter(),
check_deprecated_logger_config()
]
|> Enum.reduce(:ok, fn
:ok, :ok -> :ok
@ -415,4 +418,72 @@ defmodule Pleroma.Config.DeprecationWarnings do
:ok
end
end
defp merge_deprecated_logger_config(config) do
formatter_config = Enum.filter(config, fn {k, _} -> k in @logger_formatter_config_knobs end)
formatter = Logger.default_formatter(formatter_config)
log_level = Keyword.get(config, :level, nil)
Logger.debug("""
Reconfiguring console Logger with deprecated configuration syntax.
Handler configuration:
log_level: #{inspect(log_level)}
Formatter:
#{inspect(formatter)}
""")
if Config.get(:env) != :test do
if log_level, do: :logger.update_handler_config(:default, :level, log_level)
:logger.set_handler_config(:default, :formatter, formatter)
end
end
@spec check_deprecated_logger_config() :: :ok | :error
def check_deprecated_logger_config do
backends_config = Application.get_env(:logger, :backends)
console_config = Application.get_env(:logger, :console)
# NOTE: No need to merge the old backends config since it still works.
# And new configuration will just add new Logger backends.
backend =
if backends_config do
Logger.warning("""
!!!DEPRECATION WARNING!!!
Your configuration is using deprecated syntax for configuring backends of Elixir's logger.
`config :logger, backends: [...]` is deprecated syntax due to changes in Elixir.
Please update your configuration at your earliest convenience to use:
`config :pleroma, :logger, backends: [...]`
Note: `:console` is no longer considered a backend and is used by default, you can disable it using:
`config :logger, :default_handler: false`
""")
true
else
false
end
console =
if console_config do
Logger.warning("""
!!!DEPRECATION WARNING!!!
Your configuration is using deprecated syntax for configuring logging to console.
`config :logger, :console` is deprecated syntax due to changes in Elixir.
Please update your configuration at your earliest convenience to use
`config :logger, default_handler` and `config :logger, :default_formatter`.
Note: `:default_handler` is used only for the `level` setting. All other configurations go under
`:default_formatter`. For more info visit: https://hexdocs.pm/logger/Logger.html#module-backends-and-backwards-compatibility
""")
merge_deprecated_logger_config(console_config)
true
else
false
end
if backend or console, do: :error, else: :ok
end
end

View file

@ -51,7 +51,7 @@ defmodule Pleroma.Config.TransferTask do
started_applications = Application.started_applications()
reject = [nil, :postgrex]
reject = [nil, :postgrex, :logger]
reject =
if restart_pleroma? do

View file

@ -12,13 +12,27 @@ defmodule Pleroma.Docs.Generator do
@spec list_behaviour_implementations(behaviour :: module()) :: [module()]
def list_behaviour_implementations(behaviour) do
:code.all_loaded()
|> Enum.filter(fn {module, _} ->
# :code.all_loaded() no longer works on Elixir 1.19 and newer since modules are lazily loaded.
# We can get a complete list of modules by interating over Application.loaded_applications/0,
# but that includes every module which would get loaded below, which is very slow.
# Hardcode the Pleroma application instead, which is good enough for now.
application_modules =
case :application.get_key(:pleroma, :modules) do
{:ok, modules} -> modules
:undefined -> []
end
# At compile time or in tests some modules will be outside of the Pleroma application.
loaded_modules = Enum.map(:code.all_loaded(), fn {module, _} -> module end)
(application_modules ++ loaded_modules)
|> Enum.uniq()
|> Enum.filter(fn module ->
Code.ensure_loaded(module)
# This shouldn't be needed as all modules are expected to have module_info/1,
# but in test environments some transient modules `:elixir_compiler_XX`
# are loaded for some reason (where XX is a random integer).
Code.ensure_loaded(module)
if function_exported?(module, :module_info, 1) do
module.module_info(:attributes)
|> Keyword.get_values(:behaviour)
@ -26,7 +40,6 @@ defmodule Pleroma.Docs.Generator do
|> Enum.member?(behaviour)
end
end)
|> Enum.map(fn {module, _} -> module end)
end
@doc """

View file

@ -6,9 +6,12 @@ defmodule Pleroma.Gun do
@callback open(charlist(), pos_integer(), map()) :: {:ok, pid()}
@callback info(pid()) :: map()
@callback close(pid()) :: :ok
@callback cancel(pid(), reference() | [reference()]) :: :ok
@callback await_up(pid, pos_integer()) :: {:ok, atom()} | {:error, atom()}
@callback connect(pid(), map()) :: reference()
@callback await(pid(), reference()) :: {:response, :fin, 200, []}
@callback await(pid(), reference()) :: tuple()
@callback await_tunnel_up(pid(), reference() | :undefined, timeout()) ::
{:ok, atom()} | {:error, term()}
@callback set_owner(pid(), pid()) :: :ok
defp api, do: Pleroma.Config.get([Pleroma.Gun], Pleroma.Gun.API)
@ -19,11 +22,15 @@ defmodule Pleroma.Gun do
def close(pid), do: api().close(pid)
def cancel(pid, stream), do: api().cancel(pid, stream)
def await_up(pid, timeout \\ 5_000), do: api().await_up(pid, timeout)
def connect(pid, opts), do: api().connect(pid, opts)
def await(pid, ref), do: api().await(pid, ref)
def await_tunnel_up(pid, ref, timeout), do: api().await_tunnel_up(pid, ref, timeout)
def set_owner(pid, owner), do: api().set_owner(pid, owner)
end

View file

@ -32,6 +32,9 @@ defmodule Pleroma.Gun.API do
@impl Gun
defdelegate close(pid), to: :gun
@impl Gun
defdelegate cancel(pid, stream), to: :gun
@impl Gun
defdelegate await_up(pid, timeout \\ 5_000), to: :gun
@ -41,6 +44,28 @@ defmodule Pleroma.Gun.API do
@impl Gun
defdelegate await(pid, ref), to: :gun
@impl Gun
def await_tunnel_up(pid, stream_ref, timeout) do
monitor_ref = Process.monitor(pid)
result =
receive do
{:gun_tunnel_up, ^pid, ^stream_ref, protocol} ->
{:ok, protocol}
{:gun_down, ^pid, _protocol, reason, _killed_streams} ->
{:error, {:down, reason}}
{:DOWN, ^monitor_ref, :process, ^pid, reason} ->
{:error, {:down, reason}}
after
timeout -> {:error, :timeout}
end
Process.demonitor(monitor_ref, [:flush])
result
end
@impl Gun
defdelegate set_owner(pid, owner), to: :gun
end

View file

@ -42,18 +42,34 @@ defmodule Pleroma.Gun.Conn do
Map.put(opts, :tls_opts, tls_opts)
end
defp do_open(%URI{scheme: "http"} = uri, %{proxy: {proxy_host, proxy_port}} = opts) do
with open_opts <- opts |> proxy_open_opts() |> Map.put(:protocols, [:http]),
{:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
{:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
{:ok, conn, protocol, :forward_proxy}
else
error ->
Logger.warning(
"Opening forward proxy connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
)
error
end
end
defp do_open(uri, %{proxy: {proxy_host, proxy_port}} = opts) do
connect_opts =
uri
|> destination_opts()
|> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
|> add_proxy_auth(opts)
with open_opts <- Map.delete(opts, :tls_opts),
with open_opts <- proxy_open_opts(opts),
{:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
{:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]),
stream <- Gun.connect(conn, connect_opts),
{:response, :fin, 200, _} <- Gun.await(conn, stream) do
{:ok, conn, protocol}
{:ok, _proxy_protocol} <- await_up(conn, opts[:connect_timeout]) do
conn
|> Gun.connect(connect_opts)
|> await_proxy_connect(conn, opts[:connect_timeout])
else
error ->
Logger.warning(
@ -64,30 +80,29 @@ defmodule Pleroma.Gun.Conn do
end
end
defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts) do
version =
proxy_type
|> to_string()
|> String.last()
|> case do
"4" -> 4
_ -> 5
end
defp do_open(_uri, %{proxy: {:socks4, _proxy_host, _proxy_port}}) do
{:error, :socks4_unsupported}
end
defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts)
when proxy_type in [:socks, :socks5] do
socks_opts =
uri
|> destination_opts()
|> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
|> Map.put(:version, version)
|> Map.put(:version, 5)
|> add_socks_proxy_auth(opts)
opts =
open_opts =
opts
|> proxy_open_opts()
|> Map.put(:protocols, [:socks])
|> Map.put(:socks_opts, socks_opts)
with {:ok, conn} <- Gun.open(proxy_host, proxy_port, opts),
{:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]) do
{:ok, conn, protocol}
with {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
{:ok, :socks} <- await_up(conn, opts[:connect_timeout]),
{:ok, protocol} <- await_tunnel_up(conn, :undefined, opts[:connect_timeout]) do
{:ok, conn, protocol, nil}
else
error ->
Logger.warning(
@ -100,10 +115,11 @@ defmodule Pleroma.Gun.Conn do
defp do_open(%URI{host: host, port: port} = uri, opts) do
host = Pleroma.HTTP.AdapterHelper.parse_host(host)
opts = Map.put(opts, :transport, transport(uri.scheme))
with {:ok, conn} <- Gun.open(host, port, opts),
{:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]) do
{:ok, conn, protocol}
{:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
{:ok, conn, protocol, nil}
else
error ->
Logger.warning(
@ -120,11 +136,90 @@ defmodule Pleroma.Gun.Conn do
end
defp add_http2_opts(opts, "https", tls_opts) do
Map.merge(opts, %{protocols: [:http2], transport: :tls, tls_opts: tls_opts})
Map.merge(opts, %{protocols: [:http2, :http], transport: :tls, tls_opts: tls_opts})
end
defp add_http2_opts(opts, _, _), do: opts
defp add_proxy_auth(connect_opts, %{proxy_auth: {username, password}})
when is_binary(username) and is_binary(password) do
Map.merge(connect_opts, %{username: username, password: password})
end
defp add_proxy_auth(connect_opts, _), do: connect_opts
defp add_socks_proxy_auth(socks_opts, %{proxy_auth: {username, password}})
when is_binary(username) and is_binary(password) do
Map.put(socks_opts, :auth, [{:username_password, username, password}])
end
defp add_socks_proxy_auth(socks_opts, _), do: socks_opts
defp proxy_open_opts(opts) do
proxy_tls_opts = Map.get(opts, :proxy_tls_opts, [])
opts =
opts
|> Map.drop([:proxy, :proxy_auth, :proxy_tls_opts, :tls_opts])
|> Map.put_new(:transport, :tcp)
if opts.transport == :tls do
Map.put(opts, :tls_opts, proxy_tls_opts)
else
opts
end
end
defp await_up(conn, timeout) do
case Gun.await_up(conn, timeout) do
{:ok, protocol} ->
{:ok, protocol}
error ->
Gun.close(conn)
error
end
end
defp await_proxy_connect(stream, conn, timeout) do
case Gun.await(conn, stream) do
{:response, :fin, 200, _headers} ->
case await_tunnel_up(conn, stream, timeout) do
{:ok, protocol} -> {:ok, conn, protocol, stream}
error -> error
end
{:response, _fin, 407, _headers} ->
close_with_error(conn, :proxy_auth_failed)
{:response, _fin, status, _headers} when is_integer(status) ->
close_with_error(conn, {:proxy_connect_failed, status})
error ->
Gun.close(conn)
error
end
end
defp close_with_error(conn, reason) do
Gun.close(conn)
{:error, reason}
end
defp transport("https"), do: :tls
defp transport(_), do: :tcp
defp await_tunnel_up(conn, stream_ref, timeout) do
case Gun.await_tunnel_up(conn, stream_ref, timeout) do
{:ok, protocol} ->
{:ok, protocol}
error ->
Gun.close(conn)
error
end
end
def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
"#{scheme}://#{host}#{path}"
end

View file

@ -4,53 +4,132 @@
defmodule Pleroma.Gun.ConnectionPool do
@registry __MODULE__
@locks Module.concat(__MODULE__, Locks)
alias Pleroma.Gun.ConnectionPool.WorkerSupervisor
@connection_option_keys [
:http_opts,
:http2_opts,
:protocols,
:proxy,
:proxy_auth,
:proxy_tls_opts,
:retry,
:retry_timeout,
:socks_opts,
:tcp_opts,
:tls_opts,
:transport,
:ws_opts
]
def children do
[
{Registry, keys: :unique, name: @registry},
{Registry, keys: :duplicate, name: @registry},
{Registry, keys: :unique, name: @locks},
Pleroma.Gun.ConnectionPool.WorkerSupervisor
]
end
@spec get_conn(URI.t(), keyword()) :: {:ok, pid()} | {:error, term()}
def get_conn(uri, opts) do
key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
key = connection_key(uri, opts)
case Registry.lookup(@registry, key) do
# The key has already been registered, but connection is not up yet
[{worker_pid, nil}] ->
get_gun_pid_from_worker(worker_pid, true)
[{worker_pid, {gun_pid, _used_by, _crf, _last_reference}}] ->
GenServer.call(worker_pid, :add_client)
case checkout_existing(Registry.lookup(@registry, key)) do
{:ok, gun_pid} ->
{:ok, gun_pid}
[] ->
# :gun.set_owner fails in :connected state for whatevever reason,
# so we open the connection in the process directly and send it's pid back
# We trust gun to handle timeouts by itself
case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
{:ok, worker_pid} ->
get_gun_pid_from_worker(worker_pid, false)
:not_available ->
provision_or_wait(key, uri, opts)
end
end
{:error, {:already_started, worker_pid}} ->
get_gun_pid_from_worker(worker_pid, true)
defp checkout_existing([]), do: :not_available
err ->
err
defp checkout_existing([{_worker_pid, nil} | workers]), do: checkout_existing(workers)
defp checkout_existing([{worker_pid, _value} | workers]) do
case checkout_worker(worker_pid) do
{:ok, gun_pid} -> {:ok, gun_pid}
_ -> checkout_existing(workers)
end
end
defp provision_or_wait(key, uri, opts) do
result = with_route_lock(key, fn -> provision(key, uri, opts) end)
case result do
{:ok, gun_pid} ->
{:ok, gun_pid}
{:new, worker_pid} ->
get_gun_pid_from_worker(worker_pid)
{:wait, worker_pid} ->
case checkout_worker(worker_pid, :infinity) do
{:ok, gun_pid} -> {:ok, gun_pid}
:busy -> get_conn(uri, opts)
error -> error
end
error ->
error
end
end
defp provision(key, uri, opts) do
workers = Registry.lookup(@registry, key)
case checkout_existing(workers) do
{:ok, gun_pid} ->
{:ok, gun_pid}
:not_available ->
case Enum.find(workers, fn {_worker_pid, value} -> is_nil(value) end) do
{worker_pid, nil} -> {:wait, worker_pid}
nil -> start_worker(key, uri, opts)
end
end
end
defp get_gun_pid_from_worker(worker_pid, register) do
defp with_route_lock(key, function) do
case Registry.register(@locks, key, nil) do
{:ok, _} ->
try do
function.()
after
Registry.unregister(@locks, key)
end
{:error, {:already_registered, _owner}} ->
Process.sleep(1)
with_route_lock(key, function)
end
end
defp start_worker(key, uri, opts) do
# :gun.set_owner fails in :connected state, so the worker opens the connection itself.
case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
{:ok, worker_pid} -> {:new, worker_pid}
error -> error
end
end
defp checkout_worker(worker_pid, timeout \\ 5_000) do
try do
GenServer.call(worker_pid, :checkout, timeout)
catch
:exit, reason -> {:error, {:connection_worker_exit, reason}}
end
end
defp get_gun_pid_from_worker(worker_pid) do
# GenServer.call will block the process for timeout length if
# the server crashes on startup (which will happen if gun fails to connect)
# so instead we use cast + monitor
ref = Process.monitor(worker_pid)
if register, do: GenServer.cast(worker_pid, {:add_client, self()})
receive do
{:conn_pid, pid} ->
@ -68,19 +147,115 @@ defmodule Pleroma.Gun.ConnectionPool do
@spec release_conn(pid()) :: :ok
def release_conn(conn_pid) do
# :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _, _, _}}} when gun_pid == conn_pid ->
# :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _tunnel}}} when gun_pid == conn_pid ->
# worker_pid end)
query_result =
Registry.select(@registry, [
{{:_, :"$1", {:"$2", :_, :_, :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
{{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
])
case query_result do
[worker_pid] ->
GenServer.call(worker_pid, :remove_client)
try do
GenServer.call(worker_pid, :remove_client)
catch
:exit, _ -> :ok
end
[] ->
:ok
end
end
@spec discard_conn(pid()) :: :ok
def discard_conn(conn_pid) do
case worker_for_conn(conn_pid) do
nil ->
:ok
worker_pid ->
DynamicSupervisor.terminate_child(WorkerSupervisor, worker_pid)
end
end
@spec cancel_stream(pid(), reference() | [reference()]) :: :ok
def cancel_stream(conn_pid, stream) do
case worker_for_conn(conn_pid) do
nil ->
Pleroma.Gun.cancel(conn_pid, stream)
worker_pid ->
try do
GenServer.call(worker_pid, {:cancel_stream, stream})
catch
:exit, _ -> :ok
end
end
end
@spec register_stream(pid(), reference() | [reference()]) :: :ok
def register_stream(conn_pid, stream) do
call_worker(conn_pid, {:register_stream, stream})
end
@spec finish_stream(pid(), reference() | [reference()]) :: :ok
def finish_stream(conn_pid, stream) do
call_worker(conn_pid, {:finish_stream, stream})
end
@spec release_stream(pid(), reference() | [reference()]) :: :ok
def release_stream(conn_pid, stream) do
call_worker(conn_pid, {:release_stream, stream})
end
@spec tunnel_ref(pid()) :: reference() | :forward_proxy | nil
def tunnel_ref(conn_pid) do
case Registry.select(@registry, [
{{:_, :_, {:"$1", :"$2"}}, [{:==, :"$1", conn_pid}], [:"$2"]}
]) do
[tunnel] -> tunnel
[] -> nil
end
end
defp worker_for_conn(conn_pid) do
case Registry.select(@registry, [
{{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
]) do
[worker_pid] -> worker_pid
[] -> nil
end
end
defp call_worker(conn_pid, message) do
case worker_for_conn(conn_pid) do
nil ->
:ok
worker_pid ->
try do
GenServer.call(worker_pid, message)
catch
:exit, _ -> :ok
end
end
end
defp connection_key(uri, opts) do
origin = {
String.downcase(uri.scheme),
String.downcase(uri.host),
uri.port
}
profile =
opts
|> Map.new()
|> Map.take(@connection_option_keys)
|> :erlang.term_to_binary()
|> then(&:crypto.hash(:sha256, &1))
fingerprint = Base.encode16(profile, case: :lower)
Enum.join(Tuple.to_list(origin) ++ [fingerprint], ":")
end
end

View file

@ -9,11 +9,11 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
def start_monitor do
pid =
case GenServer.start(__MODULE__, [], name: {:via, Registry, {registry(), "reclaimer"}}) do
case GenServer.start(__MODULE__, [], name: __MODULE__) do
{:ok, pid} ->
pid
{:error, {:already_registered, pid}} ->
{:error, {:already_started, pid}} ->
pid
end
@ -41,17 +41,16 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
reclaim_max: reclaim_max
})
# :ets.fun2ms(
# fn {_, {worker_pid, {_, used_by, crf, last_reference}}} when used_by == [] ->
# {worker_pid, crf, last_reference} end)
unused_conns =
workers =
Registry.select(
registry(),
[
{{:_, :"$1", {:_, :"$2", :"$3", :"$4"}}, [{:==, :"$2", []}], [{{:"$1", :"$3", :"$4"}}]}
{{:_, :"$1", {:_, :_}}, [], [:"$1"]}
]
)
unused_conns = Enum.flat_map(workers, &reclaim_info/1)
case unused_conns do
[] ->
:telemetry.execute(
@ -65,25 +64,43 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
{:stop, :no_unused_conns, nil}
unused_conns ->
reclaimed =
reclaimed_count =
unused_conns
|> Enum.sort(fn {_pid1, crf1, last_reference1}, {_pid2, crf2, last_reference2} ->
crf1 <= crf2 and last_reference1 <= last_reference2
end)
|> Enum.take(reclaim_max)
|> Enum.sort_by(fn {_pid, crf, last_reference} -> {crf, last_reference} end)
|> Enum.reduce_while(0, fn
_worker, count when count == reclaim_max ->
{:halt, count}
reclaimed
|> Enum.each(fn {pid, _, _} ->
DynamicSupervisor.terminate_child(Pleroma.Gun.ConnectionPool.WorkerSupervisor, pid)
end)
{worker_pid, _crf, _last_reference}, count ->
if reclaim(worker_pid), do: {:cont, count + 1}, else: {:cont, count}
end)
:telemetry.execute(
[:pleroma, :connection_pool, :reclaim, :stop],
%{reclaimed_count: Enum.count(reclaimed)},
%{reclaimed_count: reclaimed_count},
%{max_connections: max_connections}
)
{:stop, :normal, nil}
end
end
defp reclaim_info(worker_pid) do
try do
case GenServer.call(worker_pid, :reclaim_info) do
{:idle, crf, last_reference} -> [{worker_pid, crf, last_reference}]
:in_use -> []
end
catch
:exit, _ -> []
end
end
defp reclaim(worker_pid) do
try do
GenServer.call(worker_pid, :reclaim) == :reclaimed
catch
:exit, _ -> false
end
end
end

View file

@ -8,25 +8,22 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
defp registry, do: Pleroma.Gun.ConnectionPool
def start_link([key | _] = opts) do
GenServer.start_link(__MODULE__, opts, name: {:via, Registry, {registry(), key}})
end
def start_link(opts), do: GenServer.start_link(__MODULE__, opts)
@impl true
def init([_key, _uri, _opts, _client_pid] = opts) do
def init([key, _uri, _opts, _client_pid] = opts) do
{:ok, _} = Registry.register(registry(), key, nil)
{:ok, nil, {:continue, {:connect, opts}}}
end
@impl true
def handle_continue({:connect, [key, uri, opts, client_pid]}, _) do
with {:ok, conn_pid, protocol} <- Gun.Conn.open(uri, opts),
with {:ok, conn_pid, protocol, tunnel} <- Gun.Conn.open(uri, opts),
Process.link(conn_pid) do
time = :erlang.monotonic_time(:millisecond)
{_, _} =
Registry.update_value(registry(), key, fn _ ->
{conn_pid, [client_pid], 1, time}
end)
Registry.unregister(registry(), key)
{:ok, _} = Registry.register(registry(), key, {conn_pid, tunnel})
send(client_pid, {:conn_pid, conn_pid})
@ -34,8 +31,14 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
%{
key: key,
timer: nil,
client_monitors: %{client_pid => Process.monitor(client_pid)},
protocol: protocol
clients: %{
client_pid => %{count: 1, monitor: Process.monitor(client_pid), streams: MapSet.new()}
},
conn_pid: conn_pid,
protocol: protocol,
proxied: Keyword.has_key?(opts, :proxy),
crf: 1,
last_reference: time
}, :hibernate}
else
err ->
@ -44,106 +47,225 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
end
@impl true
def handle_cast({:add_client, client_pid}, state) do
case handle_call(:add_client, {client_pid, nil}, state) do
{:reply, conn_pid, state, :hibernate} ->
send(client_pid, {:conn_pid, conn_pid})
{:noreply, state, :hibernate}
end
def handle_call(:checkout, _from, %{protocol: protocol, clients: clients} = state)
when protocol != :http2 and map_size(clients) > 0 do
{:reply, :busy, state, :hibernate}
end
@impl true
def handle_cast({:remove_client, client_pid}, state) do
case handle_call(:remove_client, {client_pid, nil}, state) do
{:reply, _, state, :hibernate} ->
{:noreply, state, :hibernate}
end
end
@impl true
def handle_call(:add_client, {client_pid, _}, %{key: key, protocol: protocol} = state) do
def handle_call(:checkout, {client_pid, _}, %{key: key, protocol: protocol} = state) do
time = :erlang.monotonic_time(:millisecond)
{{conn_pid, used_by, _, _}, _} =
Registry.update_value(registry(), key, fn {conn_pid, used_by, crf, last_reference} ->
{conn_pid, [client_pid | used_by], crf(time - last_reference, crf), time}
end)
[{_, {conn_pid, _tunnel}}] =
Registry.lookup(registry(), key) |> Enum.filter(&(elem(&1, 0) == self()))
state =
state
|> cancel_idle_timer()
|> add_client(client_pid)
|> Map.put(:crf, crf(time - state.last_reference, state.crf))
|> Map.put(:last_reference, time)
:telemetry.execute(
[:pleroma, :connection_pool, :client, :add],
%{client_pid: client_pid, clients: used_by},
%{client_pid: client_pid, clients: Map.keys(state.clients)},
%{key: state.key, protocol: protocol}
)
state =
if state.timer != nil do
Process.cancel_timer(state[:timer])
%{state | timer: nil}
else
state
end
ref = Process.monitor(client_pid)
state = put_in(state.client_monitors[client_pid], ref)
{:reply, conn_pid, state, :hibernate}
{:reply, {:ok, conn_pid}, state, :hibernate}
end
@impl true
def handle_call(:remove_client, {client_pid, _}, %{key: key} = state) do
{{_conn_pid, used_by, _crf, _last_reference}, _} =
Registry.update_value(registry(), key, fn {conn_pid, used_by, crf, last_reference} ->
{conn_pid, List.delete(used_by, client_pid), crf, last_reference}
end)
def handle_call(:remove_client, {client_pid, _}, state) do
case active_streams(state, client_pid) do
[] ->
{:reply, :ok, remove_client(state, client_pid, true), :hibernate}
{ref, state} = pop_in(state.client_monitors[client_pid])
streams when state.protocol == :http2 ->
Enum.each(streams, &(:ok = Gun.cancel(state.conn_pid, &1)))
{:reply, :ok, remove_client(state, client_pid, true), :hibernate}
Process.demonitor(ref, [:flush])
_streams ->
{:stop, :normal, :ok, state}
end
end
timer =
if used_by == [] do
max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
Process.send_after(self(), :idle_close, max_idle)
else
nil
end
def handle_call({:register_stream, stream}, {client_pid, _}, state) do
{:reply, :ok, update_stream(state, client_pid, &MapSet.put(&1, stream)), :hibernate}
end
{:reply, :ok, %{state | timer: timer}, :hibernate}
def handle_call({:finish_stream, stream}, {client_pid, _}, state) do
{:reply, :ok, update_stream(state, client_pid, &MapSet.delete(&1, stream)), :hibernate}
end
def handle_call({:release_stream, stream}, {client_pid, _}, state) do
state = update_stream(state, client_pid, &MapSet.delete(&1, stream))
{:reply, :ok, remove_client(state, client_pid, true), :hibernate}
end
def handle_call({:cancel_stream, stream}, {client_pid, _}, %{protocol: :http2} = state) do
:ok = Gun.cancel(state.conn_pid, stream)
state = update_stream(state, client_pid, &MapSet.delete(&1, stream))
{:reply, :ok, remove_client(state, client_pid, true), :hibernate}
end
def handle_call({:cancel_stream, _stream}, _from, state) do
{:stop, :normal, :ok, state}
end
def handle_call(:reclaim_info, _from, %{clients: clients} = state)
when map_size(clients) == 0 do
{:reply, {:idle, state.crf, state.last_reference}, state, :hibernate}
end
def handle_call(:reclaim_info, _from, state) do
{:reply, :in_use, state, :hibernate}
end
def handle_call(:reclaim, _from, %{clients: clients} = state) when map_size(clients) == 0 do
{:stop, :normal, :reclaimed, state}
end
def handle_call(:reclaim, _from, state) do
{:reply, :in_use, state, :hibernate}
end
@impl true
def handle_info(:idle_close, state) do
# Gun monitors the owner process, and will close the connection automatically
# when it's terminated
def handle_info({:idle_close, token}, %{timer: {_timer_ref, token}, clients: clients} = state)
when map_size(clients) == 0 do
# Gun monitors the owner process, and will close the connection automatically.
{:stop, :normal, state}
end
def handle_info({:idle_close, _token}, state), do: {:noreply, state, :hibernate}
@impl true
def handle_info({:gun_up, _pid, _protocol}, state) do
{:noreply, state, :hibernate}
def handle_info({:gun_up, _pid, protocol}, state) do
{:noreply, %{state | protocol: protocol}, :hibernate}
end
# Gracefully shutdown if the connection got closed without any streams left
@impl true
def handle_info({:gun_error, conn_pid, stream, _reason}, %{conn_pid: conn_pid} = state) do
{:noreply, delete_stream(state, stream), :hibernate}
end
@impl true
def handle_info({:gun_down, _pid, _protocol, _reason, _streams}, %{proxied: true} = state) do
{:stop, :normal, state}
end
# Gracefully shutdown if the connection got closed without any streams left.
@impl true
def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
{:stop, :normal, state}
end
# Otherwise, wait for retry
# Otherwise, wait for retry.
@impl true
def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams}, state) do
{:noreply, state, :hibernate}
end
@impl true
def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
:telemetry.execute(
[:pleroma, :connection_pool, :client, :dead],
%{client_pid: pid, reason: reason},
%{key: state.key}
)
def handle_info({:DOWN, ref, :process, pid, reason}, state) do
case state.clients[pid] do
%{monitor: ^ref} ->
:telemetry.execute(
[:pleroma, :connection_pool, :client, :dead],
%{client_pid: pid, reason: reason},
%{key: state.key}
)
handle_cast({:remove_client, pid}, state)
handle_dead_client(state, pid)
_ ->
{:noreply, state, :hibernate}
end
end
defp add_client(state, client_pid) do
update_in(state.clients, fn clients ->
case clients[client_pid] do
nil ->
Map.put(clients, client_pid, %{
count: 1,
monitor: Process.monitor(client_pid),
streams: MapSet.new()
})
client ->
Map.put(clients, client_pid, %{client | count: client.count + 1})
end
end)
end
defp handle_dead_client(state, client_pid) do
streams = active_streams(state, client_pid)
cond do
streams == [] ->
{:noreply, remove_client(state, client_pid, false, true), :hibernate}
state.protocol == :http2 ->
Enum.each(streams, &(:ok = Gun.cancel(state.conn_pid, &1)))
{:noreply, remove_client(state, client_pid, false, true), :hibernate}
true ->
{:stop, :normal, state}
end
end
defp active_streams(state, client_pid) do
case state.clients[client_pid] do
%{streams: streams} -> MapSet.to_list(streams)
nil -> []
end
end
defp update_stream(state, client_pid, update) do
update_in(state.clients, fn clients ->
case clients[client_pid] do
nil -> clients
client -> Map.put(clients, client_pid, %{client | streams: update.(client.streams)})
end
end)
end
defp delete_stream(state, stream) do
update_in(state.clients, fn clients ->
Map.new(clients, fn {pid, client} ->
{pid, %{client | streams: MapSet.delete(client.streams, stream)}}
end)
end)
end
defp remove_client(state, client_pid, demonitor, all \\ false) do
case state.clients[client_pid] do
nil ->
state
%{count: count} = client when count > 1 and not all ->
put_in(state.clients[client_pid], %{client | count: count - 1})
client ->
if demonitor, do: Process.demonitor(client.monitor, [:flush])
state = update_in(state.clients, &Map.delete(&1, client_pid))
if map_size(state.clients) == 0, do: schedule_idle_close(state), else: state
end
end
defp cancel_idle_timer(%{timer: nil} = state), do: state
defp cancel_idle_timer(%{timer: {timer_ref, _token}} = state) do
Process.cancel_timer(timer_ref)
%{state | timer: nil}
end
defp schedule_idle_close(state) do
max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
token = make_ref()
timer_ref = Process.send_after(self(), {:idle_close, token}, max_idle)
%{state | timer: {timer_ref, token}}
end
# LRFU policy: https://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.55.1478

View file

@ -26,7 +26,14 @@ defmodule Pleroma.Gun.ConnectionPool.WorkerSupervisor do
def start_worker(opts, true) do
case DynamicSupervisor.start_child(__MODULE__, {Worker, opts}) do
{:error, :max_children} ->
:telemetry.execute([:pleroma, :connection_pool, :provision_failure], %{opts: opts})
[key | _] = opts
:telemetry.execute(
[:pleroma, :connection_pool, :provision_failure],
%{},
%{key: key}
)
{:error, :pool_full}
res ->

View file

@ -0,0 +1,253 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.HTTP.Adapter.Gun do
@behaviour Tesla.Adapter
alias Pleroma.Gun.ConnectionPool
alias Tesla.Multipart
@default_timeout 1_000
@impl Tesla.Adapter
def call(env, opts) do
adapter_opts =
Tesla.Adapter.opts(
[close_conn: true, body_as: :plain, send_body: :at_once, receive: true],
env,
opts
)
|> Map.new()
with {:ok, status, headers, body} <- request(env, adapter_opts) do
env = %{env | status: status, headers: format_headers(headers), body: body}
{:ok, put_stream_ref(env, body)}
end
end
defp put_stream_ref(env, %{stream: stream}) do
put_in(env.opts[:adapter][:stream], stream)
end
defp put_stream_ref(env, _body), do: env
defp request(env, %{conn: conn, tunnel: tunnel} = opts) do
uri = URI.parse(Tesla.build_url(env))
path = request_path(uri, tunnel)
method = Tesla.Adapter.Shared.format_method(env.method)
{headers, body, send_body} = prepare_body(env.headers, env.body, opts[:send_body])
headers = maybe_add_forward_proxy_headers(headers, uri, opts, tunnel)
request_opts = request_opts(opts[:reply_to], tunnel)
stream =
try do
open_stream(conn, method, path, headers, body, request_opts, send_body)
catch
kind, reason ->
ConnectionPool.release_conn(conn)
:erlang.raise(kind, reason, __STACKTRACE__)
end
:ok = ConnectionPool.register_stream(conn, stream)
:ok = send_stream_body(conn, stream, body, send_body)
response = read_response(conn, stream, opts)
case response do
{:error, reason} ->
ConnectionPool.cancel_stream(conn, stream)
{:error, {:gun_stream_error, reason}}
response ->
unless opts[:body_as] in [:chunks, :stream] and stream_body?(response) do
:ok = ConnectionPool.finish_stream(conn, stream)
end
if opts[:close_conn] and opts[:body_as] not in [:stream, :chunks] do
:gun.close(conn)
end
response
end
end
defp stream_body?({:ok, _status, _headers, %{stream: _stream}}), do: true
defp stream_body?({:ok, _status, _headers, %Stream{}}), do: true
defp stream_body?(_response), do: false
defp request_opts(reply_to, nil), do: %{reply_to: reply_to || self()}
defp request_opts(reply_to, :forward_proxy), do: %{reply_to: reply_to || self()}
defp request_opts(reply_to, tunnel) do
%{reply_to: reply_to || self(), tunnel: tunnel}
end
defp request_path(uri, :forward_proxy) do
path = Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
"#{uri.scheme}://#{authority(uri)}#{path}"
end
defp request_path(uri, _tunnel) do
Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
end
defp maybe_add_forward_proxy_headers(headers, uri, opts, :forward_proxy) do
headers
|> put_header_new("host", authority(uri))
|> maybe_put_proxy_authorization(opts)
end
defp maybe_add_forward_proxy_headers(headers, _uri, _opts, _tunnel), do: headers
defp maybe_put_proxy_authorization(headers, %{proxy_auth: {username, password}})
when is_binary(username) and is_binary(password) do
value = "Basic " <> Base.encode64(username <> ":" <> password)
List.keystore(headers, "proxy-authorization", 0, {"proxy-authorization", value})
end
defp maybe_put_proxy_authorization(headers, _opts), do: headers
defp put_header_new(headers, name, value) do
if List.keymember?(headers, name, 0), do: headers, else: [{name, value} | headers]
end
defp authority(%URI{scheme: scheme, host: host, port: port}) do
host = if String.contains?(host, ":"), do: "[#{host}]", else: host
if port == URI.default_port(scheme), do: host, else: "#{host}:#{port}"
end
defp prepare_body(headers, %Multipart{} = multipart, _send_body) do
{format_headers(headers ++ Multipart.headers(multipart)), Multipart.body(multipart), :stream}
end
defp prepare_body(headers, body, _send_body)
when is_struct(body, Stream) or is_function(body) do
{format_headers(headers), body, :stream}
end
defp prepare_body(headers, body, send_body) do
{format_headers(headers), body || "", send_body}
end
defp open_stream(conn, method, path, headers, _body, request_opts, :stream) do
:gun.headers(conn, method, path, headers, request_opts)
end
defp open_stream(conn, method, path, headers, body, request_opts, :at_once) do
:gun.request(conn, method, path, headers, body, request_opts)
end
defp send_stream_body(conn, stream, body, :stream) do
try do
for data <- body, do: :ok = :gun.data(conn, stream, :nofin, data)
:gun.data(conn, stream, :fin, "")
catch
kind, reason ->
ConnectionPool.cancel_stream(conn, stream)
:erlang.raise(kind, reason, __STACKTRACE__)
end
end
defp send_stream_body(_conn, _stream, _body, :at_once), do: :ok
defp read_response(conn, stream, opts) do
receive? = opts[:receive]
receive do
{:gun_response, ^conn, ^stream, :fin, status, headers} ->
{:ok, status, headers, ""}
{:gun_response, ^conn, ^stream, :nofin, status, headers} ->
format_response(conn, stream, opts, status, headers, opts[:body_as])
{:gun_up, ^conn, _protocol} when receive? ->
read_response(conn, stream, opts)
{:gun_error, ^conn, reason} ->
{:error, reason}
{:gun_error, ^conn, ^stream, reason} ->
{:error, reason}
{:gun_down, ^conn, _protocol, _reason, _killed_streams} when receive? ->
read_response(conn, stream, opts)
{:DOWN, _ref, :process, ^conn, reason} ->
{:error, reason}
after
opts[:timeout] || @default_timeout -> {:error, :recv_response_timeout}
end
end
defp format_response(conn, stream, opts, status, headers, :plain) do
case read_body(conn, stream, opts) do
{:ok, body} ->
{:ok, status, headers, body}
{:error, error} ->
:ok = :gun.flush(stream)
{:error, error}
end
end
defp format_response(conn, stream, opts, status, headers, :stream) do
body =
Stream.resource(
fn -> :reading end,
fn
:reading ->
case Tesla.Adapter.Gun.read_chunk(conn, stream, opts) do
{:nofin, part} -> {[part], :reading}
{:fin, part} -> {[part], :done}
{:error, reason} -> raise "Gun stream failed: #{inspect(reason)}"
end
:done ->
{:halt, :done}
end,
fn _state ->
if opts[:close_conn], do: :gun.close(conn)
end
)
{:ok, status, headers, body}
end
defp format_response(conn, stream, opts, status, headers, :chunks) do
{:ok, status, headers, %{pid: conn, stream: stream, opts: Map.to_list(opts)}}
end
defp read_body(conn, stream, opts, acc \\ "") do
receive do
{:gun_data, ^conn, ^stream, :fin, body} ->
append_body(acc, body, opts[:max_body])
{:gun_data, ^conn, ^stream, :nofin, body} ->
with {:ok, acc} <- append_body(acc, body, opts[:max_body]) do
read_body(conn, stream, opts, acc)
end
{:gun_error, ^conn, ^stream, reason} ->
{:error, reason}
{:DOWN, _ref, :process, ^conn, reason} ->
{:error, reason}
after
opts[:timeout] || @default_timeout -> {:error, :recv_body_timeout}
end
end
defp append_body(acc, part, nil), do: {:ok, acc <> part}
defp append_body(acc, part, limit) do
body = acc <> part
if byte_size(body) <= limit, do: {:ok, body}, else: {:error, :body_too_large}
end
defp format_headers(headers) do
Enum.map(headers, fn {key, value} ->
{String.downcase(to_string(key)), to_string(value)}
end)
end
end

View file

@ -48,10 +48,10 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do
Keyword.put(opts, :timeout, recv_timeout)
end
# Gun uses [body_as: :stream]
# Keep the pool lease until the caller consumes or cancels the body.
defp maybe_stream(opts) do
case Keyword.pop(opts, :stream, nil) do
{true, opts} -> Keyword.put(opts, :body_as, :stream)
{true, opts} -> Keyword.put(opts, :body_as, :chunks)
{_, opts} -> opts
end
end

View file

@ -9,6 +9,7 @@ defmodule Pleroma.List do
import Ecto.Changeset
alias Pleroma.Activity
alias Pleroma.Emoji
alias Pleroma.Repo
alias Pleroma.User
@ -18,14 +19,53 @@ defmodule Pleroma.List do
field(:following, {:array, :string}, default: [])
field(:ap_id, :string)
field(:exclusive, :boolean, default: false)
field(:emoji, :string)
timestamps()
end
def update_changeset(list, attrs \\ %{}) do
list
|> cast(attrs, [:title, :exclusive])
|> cast(attrs, [:title, :exclusive, :emoji])
|> validate_required([:title])
|> fix_emoji()
|> validate_emoji()
end
defp fix_emoji(changeset) do
with {:emoji_field, emoji} when is_binary(emoji) <-
{:emoji_field, get_field(changeset, :emoji)},
{:fixed_emoji, emoji} <-
{:fixed_emoji,
emoji
|> Emoji.fully_qualify_emoji()
|> Emoji.maybe_quote()} do
put_change(changeset, :emoji, emoji)
else
{:emoji_field, _} -> changeset
end
end
defp validate_emoji(changeset) do
validate_change(changeset, :emoji, fn
:emoji, nil ->
[]
:emoji, emoji ->
if Emoji.unicode?(emoji) or valid_local_custom_emoji?(emoji) do
[]
else
[emoji: "Invalid emoji"]
end
end)
end
defp valid_local_custom_emoji?(emoji) do
with %{file: _path} <- Emoji.get(emoji) do
true
else
_ -> false
end
end
def follow_changeset(list, attrs \\ %{}) do

View file

@ -35,6 +35,7 @@ defmodule Pleroma.Notification do
# remember to add a migration to add it to the `notifications_type` enum
# as well.
field(:type, :string)
field(:group_key, :string)
belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
belongs_to(:activity, Activity, type: FlakeId.Ecto.CompatType)
@ -49,7 +50,7 @@ defmodule Pleroma.Notification do
|> type_from_activity()
notification
|> changeset(%{type: type})
|> changeset(%{type: type, group_key: grouped_notification_key(type, activity)})
|> Repo.update()
end
end
@ -62,6 +63,99 @@ defmodule Pleroma.Notification do
|> Repo.aggregate(:count, :id)
end
@groupable_notification_types ~w{favourite follow reblog}
@group_bucket_seconds 12 * 60 * 60
def groupable_notification_types, do: @groupable_notification_types
def group_notifications(notifications, grouped_types \\ nil) do
grouped_types = normalize_grouped_types(grouped_types)
{group_keys, grouped_notifications} =
Enum.reduce(notifications, {[], %{}}, fn notification,
{group_keys, grouped_notifications} ->
group_key = group_key(notification, grouped_types)
if Map.has_key?(grouped_notifications, group_key) do
{group_keys, Map.update!(grouped_notifications, group_key, &[notification | &1])}
else
{[group_key | group_keys], Map.put(grouped_notifications, group_key, [notification])}
end
end)
group_keys
|> Enum.reverse()
|> Enum.map(fn group_key ->
grouped_notifications
|> Map.fetch!(group_key)
|> Enum.reverse()
end)
end
def group_key(notification, grouped_types \\ nil)
def group_key(%Notification{} = notification, nil) do
group_key(notification, @groupable_notification_types)
end
def group_key(%Notification{type: type, group_key: group_key} = notification, grouped_types) do
grouped_types = normalize_grouped_types(grouped_types)
if type in @groupable_notification_types and type in grouped_types and is_binary(group_key) do
group_key
else
ungrouped_group_key(notification)
end
end
def normalize_grouped_types(nil), do: @groupable_notification_types
def normalize_grouped_types(types) when is_list(types), do: types
def normalize_grouped_types(type), do: [type]
defp ungrouped_group_key(%Notification{id: id}), do: "ungrouped-#{id}"
defp grouped_notification_key(type, activity) when type in @groupable_notification_types do
with target_id when is_binary(target_id) <- group_target_id(type, activity) do
# Mastodon uses Redis to reuse a recent bucket for rolling 12h windows. Pleroma keeps keys
# deterministic and self-contained; clients must treat group_key as opaque either way.
"#{type}-#{target_id}-#{group_time_bucket(activity)}"
end
end
defp grouped_notification_key(_type, _activity), do: nil
defp group_time_bucket(%Activity{inserted_at: inserted_at}) when not is_nil(inserted_at) do
inserted_at
|> NaiveDateTime.to_erl()
|> :calendar.datetime_to_gregorian_seconds()
|> div(@group_bucket_seconds)
end
defp group_time_bucket(_),
do: group_time_bucket(%Activity{inserted_at: NaiveDateTime.utc_now()})
defp group_target_id(type, activity) when type in ["favourite", "reblog"] do
with object_id when is_binary(object_id) <- object_id_for(activity),
%Activity{id: id} <- Activity.get_create_by_object_ap_id(object_id) do
to_string(id)
else
_ -> nil
end
end
defp group_target_id("follow", %{data: %{"object" => ap_id}}) do
case User.get_cached_by_ap_id(ap_id) do
%User{id: id} -> to_string(id)
_ -> nil
end
end
defp group_target_id(_, _), do: nil
defp object_id_for(%{data: %{"object" => %{"id" => id}}}) when is_binary(id), do: id
defp object_id_for(%{data: %{"object" => id}}) when is_binary(id), do: id
defp object_id_for(_), do: nil
@notification_types ~w{
favourite
follow
@ -79,7 +173,7 @@ defmodule Pleroma.Notification do
def changeset(%Notification{} = notification, attrs) do
notification
|> cast(attrs, [:seen, :type])
|> cast(attrs, [:seen, :type, :group_key])
|> validate_inclusion(:type, @notification_types)
end
@ -458,7 +552,8 @@ defmodule Pleroma.Notification do
user_id: user.id,
activity: activity,
seen: mark_as_read?(activity, user),
type: type
type: type,
group_key: grouped_notification_key(type, activity)
})
|> Marker.multi_set_last_read_id(user, "notifications")
|> Repo.transaction()

View file

@ -17,6 +17,7 @@ defmodule Pleroma.ReverseProxy do
@max_read_duration :timer.seconds(30)
@max_body_length :infinity
@failed_request_ttl :timer.seconds(60)
@sniff_bytes 8 * 1024
@methods ~w(GET HEAD)
@allowed_mime_types Pleroma.Config.get([Pleroma.Upload, :allowed_mime_types], [])
@ -59,7 +60,12 @@ defmodule Pleroma.ReverseProxy do
* `inline_content_types`:
* `true` will not alter `content-disposition` (up to the upstream),
* `false` will add `content-disposition: attachment` to any request,
* a list of whitelisted content types
* a list of whitelisted content types for which `content-disposition: inline`
is always set (overriding any upstream header) so the media can be embedded
in pages; the filename is derived from the content type
* `sniff_content_type` (default `false`): detects an image MIME type from the first
response chunk when the upstream type is missing or `application/octet-stream`.
* `req_headers`, `resp_headers` additional headers.
@ -69,11 +75,15 @@ defmodule Pleroma.ReverseProxy do
@default_options [pool: :media]
@inline_content_types [
"image/apng",
"image/avif",
"image/bmp",
"image/gif",
"image/jpeg",
"image/jpg",
"image/png",
"image/svg+xml",
"image/webp",
"audio/mpeg",
"audio/mp3",
"video/webm",
@ -92,6 +102,7 @@ defmodule Pleroma.ReverseProxy do
| {:req_headers, [{String.t(), String.t()}]}
| {:resp_headers, [{String.t(), String.t()}]}
| {:inline_content_types, boolean() | list(String.t())}
| {:sniff_content_type, boolean()}
| {:redirect_on_failure, boolean()}
@spec call(Plug.Conn.t(), String.t(), list(option())) :: Plug.Conn.t()
@ -110,12 +121,8 @@ defmodule Pleroma.ReverseProxy do
end
with {:ok, nil} <- @cachex.get(:failed_proxy_url_cache, url),
{:ok, code, headers, client} <- request(method, url, req_headers, client_opts),
:ok <-
header_length_constraint(
headers,
Keyword.get(opts, :max_body_length, @max_body_length)
) do
{:ok, code, headers, client} <-
request_with_constraints(method, url, req_headers, client_opts, opts) do
response(conn, client, url, code, headers, opts)
else
{:ok, true} ->
@ -170,7 +177,8 @@ defmodule Pleroma.ReverseProxy do
{:ok, code, headers} when code in @valid_resp_codes ->
{:ok, code, downcase_headers(headers)}
{:ok, code, _, _} ->
{:ok, code, _, client} ->
client().close(client)
{:error, {:invalid_http_response, code}}
{:ok, code, _} ->
@ -181,15 +189,37 @@ defmodule Pleroma.ReverseProxy do
end
end
defp request_with_constraints(method, url, headers, client_opts, opts) do
case request(method, url, headers, client_opts) do
{:ok, _code, headers, client} = response ->
case header_length_constraint(
headers,
Keyword.get(opts, :max_body_length, @max_body_length)
) do
:ok ->
response
error ->
client().close(client)
error
end
response ->
response
end
end
defp response(conn, client, url, status, headers, opts) do
Logger.debug("#{__MODULE__} #{status} #{url} #{inspect(headers)}")
{headers, client, prefetched} = maybe_prefetch_for_content_type(headers, client, opts)
result =
conn
|> put_resp_headers(build_resp_headers(headers, opts))
|> streaming_compat
|> send_chunked(status)
|> chunk_reply(client, opts)
|> chunk_reply(client, opts, prefetched)
case result do
{:ok, conn} ->
@ -209,18 +239,28 @@ defmodule Pleroma.ReverseProxy do
end
end
defp chunk_reply(conn, client, opts) do
defp chunk_reply(conn, client, opts, :none) do
chunk_reply(conn, client, opts, 0, 0)
end
defp chunk_reply(conn, client, opts, sent_so_far, duration) do
with {:ok, duration} <-
check_read_duration(
duration,
Keyword.get(opts, :max_read_duration, @max_read_duration)
defp chunk_reply(conn, _client, _opts, :done), do: {:ok, conn}
defp chunk_reply(conn, _client, _opts, {:error, error}), do: {:error, error, conn}
defp chunk_reply(conn, client, opts, {:ok, data, duration}) do
with :ok <-
body_size_constraint(
byte_size(data),
Keyword.get(opts, :max_body_length, @max_body_length)
),
{:ok, data, client} <- client().stream_body(client),
{:ok, duration} <- increase_read_duration(duration),
{:ok, conn} <- chunk(conn, data) do
chunk_reply(conn, client, opts, byte_size(data), duration)
else
{:error, error} -> {:error, error, conn}
end
end
defp chunk_reply(conn, client, opts, sent_so_far, duration) do
with {:ok, data, client, duration} <- read_chunk(client, duration, opts),
sent_so_far = sent_so_far + byte_size(data),
:ok <-
body_size_constraint(
@ -235,6 +275,66 @@ defmodule Pleroma.ReverseProxy do
end
end
defp read_chunk(client, duration, opts) do
with {:ok, timer} <-
check_read_duration(
duration,
Keyword.get(opts, :max_read_duration, @max_read_duration)
),
{:ok, data, client} <- client().stream_body(client),
{:ok, duration} <- increase_read_duration(timer) do
{:ok, data, client, duration}
end
end
defp maybe_prefetch_for_content_type(headers, client, opts) do
if Keyword.get(opts, :sniff_content_type, false) and generic_content_type?(headers) do
case read_chunk(client, 0, opts) do
{:ok, data, client, duration} ->
{maybe_put_image_content_type(headers, data), client, {:ok, data, duration}}
:done ->
{headers, client, :done}
{:error, error} ->
{headers, client, {:error, error}}
end
else
{headers, client, :none}
end
end
defp generic_content_type?(headers) do
headers
|> get_content_type()
|> String.trim()
|> String.downcase()
|> then(&(&1 in ["", "application/octet-stream"]))
end
defp maybe_put_image_content_type(headers, data) do
with false <- data == "",
{:ok, %{mime_type: "image/" <> _ = content_type}} <-
Majic.perform({:bytes, binary_part(data, 0, min(byte_size(data), @sniff_bytes))},
pool: Pleroma.MajicPool
) do
[
{"content-type", content_type}
| Enum.reject(headers, fn {key, _} -> key == "content-type" end)
]
else
_ -> headers
end
rescue
error ->
Logger.debug("#{__MODULE__}: content-type sniffing failed: #{Exception.message(error)}")
headers
catch
kind, reason ->
Logger.debug("#{__MODULE__}: content-type sniffing failed: #{kind}: #{inspect(reason)}")
headers
end
defp head_response(conn, url, code, headers, opts) do
Logger.debug("#{__MODULE__} #{code} #{url} #{inspect(headers)}")
@ -281,17 +381,34 @@ defmodule Pleroma.ReverseProxy do
|> Enum.filter(fn {k, _} -> k in @keep_req_headers end)
|> build_req_range_or_encoding_header(opts)
|> build_req_user_agent_header(opts)
|> Keyword.merge(Keyword.get(opts, :req_headers, []))
|> merge_headers(Keyword.get(opts, :req_headers, []))
|> maybe_force_identity_encoding(opts)
end
defp merge_headers(headers, extra_headers) do
Enum.reduce(extra_headers, headers, fn {key, value}, headers ->
List.keystore(headers, String.downcase(key), 0, {String.downcase(key), value})
end)
end
defp maybe_force_identity_encoding(headers, opts) do
if Keyword.get(opts, :sniff_content_type, false) do
List.keystore(headers, "accept-encoding", 0, {"accept-encoding", "identity"})
else
headers
end
end
# Disable content-encoding if any @range_headers are requested (see #1823).
defp build_req_range_or_encoding_header(headers, _opts) do
range? = Enum.any?(headers, fn {header, _} -> Enum.member?(@range_headers, header) end)
if range? && List.keymember?(headers, "accept-encoding", 0) do
List.keydelete(headers, "accept-encoding", 0)
else
headers
cond do
range? && List.keymember?(headers, "accept-encoding", 0) ->
List.keydelete(headers, "accept-encoding", 0)
true ->
headers
end
end
@ -384,9 +501,33 @@ defmodule Pleroma.ReverseProxy do
disposition = "attachment; filename=\"#{name}\""
List.keystore(headers, "content-disposition", 0, {"content-disposition", disposition})
replace_header(headers, "content-disposition", disposition)
else
headers
if opt == true do
headers
else
name = inline_filename(content_type)
disposition =
if name do
"inline; filename=\"#{name}\""
else
"inline"
end
replace_header(headers, "content-disposition", disposition)
end
end
end
defp replace_header(headers, key, value) do
[{key, value} | Enum.reject(headers, fn {header, _} -> header == key end)]
end
defp inline_filename(content_type) do
case MIME.extensions(content_type) do
[ext | _] when ext != "" -> "inline.#{ext}"
_ -> nil
end
end
@ -406,7 +547,7 @@ defmodule Pleroma.ReverseProxy do
defp header_length_constraint(_, _), do: :ok
defp body_size_constraint(size, limit) when is_integer(limit) and limit > 0 and size >= limit do
defp body_size_constraint(size, limit) when is_integer(limit) and limit > 0 and size > limit do
{:error, :body_too_large}
end
@ -421,7 +562,9 @@ defmodule Pleroma.ReverseProxy do
end
end
defp check_read_duration(_, _), do: {:ok, :no_duration_limit, :no_duration_limit}
defp check_read_duration(_, _), do: {:ok, :no_duration_limit}
defp increase_read_duration(:no_duration_limit), do: {:ok, :no_duration_limit}
defp increase_read_duration({previous_duration, started})
when is_integer(previous_duration) and is_integer(started) do

View file

@ -6,6 +6,7 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
@behaviour Pleroma.ReverseProxy.Client
alias Pleroma.Gun.ConnectionPool
alias Pleroma.HTTP.AdapterHelper
@type headers() :: [{String.t(), String.t()}]
@type status() :: pos_integer()
@ -20,6 +21,13 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
def request(method, url, headers, body, opts \\ []) do
check_adapter()
# MediaProxyController inserts MediaProxy proxy configuration into the ReverseProxy call.
# This config can be unformatted and in String representation, while this works for Hackney,
# it does not work for Tesla and the proxy needs to be reformatted.
# Gun AdapterHelper doesn't do it, since it does not overwrite proxy config.
proxy = AdapterHelper.format_proxy(opts[:proxy])
opts = if proxy, do: Keyword.put(opts, :proxy, proxy), else: opts
opts = Keyword.put(opts, :body_as, :chunks)
with {:ok, response} <-
@ -45,8 +53,8 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
@impl true
@spec stream_body(map()) ::
{:ok, binary(), map()} | {:error, atom() | String.t()} | :done | no_return()
def stream_body(%{pid: pid, fin: true}) do
ConnectionPool.release_conn(pid)
def stream_body(%{pid: pid, stream: stream, fin: true}) do
ConnectionPool.release_stream(pid, stream)
:done
end
@ -70,6 +78,10 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
@impl true
@spec close(map) :: :ok | no_return()
def close(%{pid: pid, stream: stream}) do
ConnectionPool.cancel_stream(pid, stream)
end
def close(%{pid: pid}) do
ConnectionPool.release_conn(pid)
end

View file

@ -104,7 +104,7 @@ defmodule Pleroma.Signature do
|> put_req_header("(request-target)", request_target)
|> put_req_header("@request-target", request_target)
@http_signatures_impl.validate_conn(conn)
@http_signatures_impl.validate_conn(conn) == true
end
@spec validate_signature(Plug.Conn.t()) :: boolean()

View file

@ -55,8 +55,8 @@ defmodule Pleroma.Telemetry.Logger do
def handle_event(
[:pleroma, :connection_pool, :provision_failure],
%{opts: [key | _]},
_,
%{key: key},
_
) do
Logger.debug(fn ->

View file

@ -12,25 +12,38 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
alias Pleroma.Gun.ConnectionPool
@impl Tesla.Middleware
def call(%Tesla.Env{url: url, opts: opts} = env, next, _) do
def call(%Tesla.Env{opts: opts} = env, next, middleware_opts) do
if opts[:adapter][:body_as] == :stream do
{:error, :pooled_streaming_requires_chunks}
else
do_call(env, next, middleware_opts)
end
end
defp do_call(%Tesla.Env{url: url, opts: opts} = env, next, _) do
uri = URI.parse(url)
# Avoid leaking connections when the middleware is called twice
# with body_as: :chunks. We assume only the middleware can set
# opts[:adapter][:conn]
if opts[:adapter][:conn] do
ConnectionPool.release_conn(opts[:adapter][:conn])
end
cleanup_previous_request(env)
case ConnectionPool.get_conn(uri, opts[:adapter]) do
{:ok, conn_pid} ->
adapter_opts = Keyword.merge(opts[:adapter], conn: conn_pid, close_conn: false)
tunnel = ConnectionPool.tunnel_ref(conn_pid)
adapter_opts =
opts[:adapter]
|> Keyword.drop([:conn, :stream, :tunnel])
|> Keyword.merge(conn: conn_pid, close_conn: false, tunnel: tunnel)
opts = Keyword.put(opts, :adapter, adapter_opts)
env = %{env | opts: opts}
next = use_pool_adapter(next)
case Tesla.run(env, next) do
{:ok, env} ->
unless opts[:adapter][:body_as] == :chunks do
unless opts[:adapter][:body_as] == :chunks and chunk_client?(env.body) do
ConnectionPool.release_conn(conn_pid)
{_, res} = pop_in(env.opts[:adapter][:conn])
{:ok, res}
@ -38,8 +51,11 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
{:ok, env}
end
{:error, {:gun_stream_error, reason}} ->
{:error, reason}
err ->
ConnectionPool.release_conn(conn_pid)
ConnectionPool.discard_conn(conn_pid)
err
end
@ -47,4 +63,34 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
err
end
end
defp cleanup_previous_request(%Tesla.Env{
body: %{pid: pid, stream: stream},
opts: opts
}) do
if opts[:adapter][:conn], do: ConnectionPool.cancel_stream(pid, stream)
end
defp cleanup_previous_request(%Tesla.Env{opts: opts}) do
conn = opts[:adapter][:conn]
stream = opts[:adapter][:stream]
cond do
conn && stream -> ConnectionPool.cancel_stream(conn, stream)
conn -> ConnectionPool.release_conn(conn)
true -> :ok
end
end
defp chunk_client?(%{pid: pid, stream: stream}) when is_pid(pid) and not is_nil(stream),
do: true
defp chunk_client?(_body), do: false
defp use_pool_adapter(next) do
List.update_at(next, -1, fn
{Tesla.Adapter.Gun, function, args} -> {Pleroma.HTTP.Adapter.Gun, function, args}
adapter -> adapter
end)
end
end

View file

@ -272,13 +272,13 @@ defmodule Pleroma.User do
end
def cached_blocked_users_ap_ids(user) do
@cachex.fetch!(:user_cache, "blocked_users_ap_ids:#{user.ap_id}", fn _ ->
@cachex.fetch!(:user_cache, cache_key(:blocked_users_ap_ids, user), fn _ ->
blocked_users_ap_ids(user)
end)
end
def cached_muted_users_ap_ids(user) do
@cachex.fetch!(:user_cache, "muted_users_ap_ids:#{user.ap_id}", fn _ ->
@cachex.fetch!(:user_cache, cache_key(:muted_users_ap_ids, user), fn _ ->
muted_users_ap_ids(user)
end)
end
@ -1253,13 +1253,18 @@ defmodule Pleroma.User do
get_cached_by_nickname(nickname)
end
defp cache_key(:id, %User{id: id}), do: cache_key(:id, id)
defp cache_key(:nickname, %User{nickname: nickname}), do: cache_key(:nickname, nickname)
defp cache_key(namespace, %User{ap_id: ap_id}), do: cache_key(namespace, ap_id)
defp cache_key(namespace, value), do: "#{namespace}:#{value}"
def set_cache({:ok, user}), do: set_cache(user)
def set_cache({:error, err}), do: {:error, err}
def set_cache(%User{} = user) do
@cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
@cachex.put(:user_cache, "nickname:#{user.nickname}", user)
@cachex.put(:user_cache, "friends_ap_ids:#{user.nickname}", get_user_friends_ap_ids(user))
@cachex.put(:user_cache, cache_key(:ap_id, user), user)
@cachex.put(:user_cache, cache_key(:nickname, user), user)
@cachex.put(:user_cache, cache_key(:friends_ap_ids, user), get_user_friends_ap_ids(user))
{:ok, user}
end
@ -1287,22 +1292,23 @@ defmodule Pleroma.User do
@spec get_cached_user_friends_ap_ids(User.t()) :: [String.t()]
def get_cached_user_friends_ap_ids(user) do
@cachex.fetch!(:user_cache, "friends_ap_ids:#{user.ap_id}", fn _ ->
@cachex.fetch!(:user_cache, cache_key(:friends_ap_ids, user), fn _ ->
get_user_friends_ap_ids(user)
end)
end
def invalidate_cache(user) do
@cachex.del(:user_cache, "ap_id:#{user.ap_id}")
@cachex.del(:user_cache, "nickname:#{user.nickname}")
@cachex.del(:user_cache, "friends_ap_ids:#{user.ap_id}")
@cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
@cachex.del(:user_cache, "muted_users_ap_ids:#{user.ap_id}")
@cachex.del(:user_cache, cache_key(:ap_id, user))
@cachex.del(:user_cache, cache_key(:nickname, user))
@cachex.del(:user_cache, cache_key(:id, user))
@cachex.del(:user_cache, cache_key(:friends_ap_ids, user))
@cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
@cachex.del(:user_cache, cache_key(:muted_users_ap_ids, user))
end
@spec get_cached_by_ap_id(String.t()) :: User.t() | nil
def get_cached_by_ap_id(ap_id) do
key = "ap_id:#{ap_id}"
key = cache_key(:ap_id, ap_id)
with {:ok, nil} <- @cachex.get(:user_cache, key),
user when not is_nil(user) <- get_by_ap_id(ap_id),
@ -1315,14 +1321,14 @@ defmodule Pleroma.User do
end
def get_cached_by_id(id) do
key = "id:#{id}"
key = cache_key(:id, id)
ap_id =
@cachex.fetch!(:user_cache, key, fn _ ->
user = get_by_id(id)
if user do
@cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
@cachex.put(:user_cache, cache_key(:ap_id, user), user)
{:commit, user.ap_id}
else
{:ignore, ""}
@ -1333,7 +1339,7 @@ defmodule Pleroma.User do
end
def get_cached_by_nickname(nickname) do
key = "nickname:#{nickname}"
key = cache_key(:nickname, nickname)
@cachex.fetch!(:user_cache, key, fn _ ->
case get_or_fetch_by_nickname(nickname) do
@ -1651,7 +1657,7 @@ defmodule Pleroma.User do
|> Oban.insert()
end
@cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
@cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
{:ok, Enum.filter([user_mute, user_notification_mute], & &1)}
end
@ -1661,7 +1667,7 @@ defmodule Pleroma.User do
with {:ok, user_mute} <- UserRelationship.delete_mute(muter, mutee),
{:ok, user_notification_mute} <-
UserRelationship.delete_notification_mute(muter, mutee) do
@cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
@cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
{:ok, [user_mute, user_notification_mute]}
end
end
@ -2821,7 +2827,7 @@ defmodule Pleroma.User do
{:ok, UserRelationship.t()} | {:error, Ecto.Changeset.t()}
defp add_to_block(%User{} = user, %User{} = blocked, expires_at) do
with {:ok, relationship} <- UserRelationship.create_block(user, blocked, expires_at) do
@cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
@cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
{:ok, relationship}
end
end
@ -2830,7 +2836,7 @@ defmodule Pleroma.User do
{:ok, UserRelationship.t()} | {:ok, nil} | {:error, Ecto.Changeset.t()}
defp remove_from_block(%User{} = user, %User{} = blocked) do
with {:ok, relationship} <- UserRelationship.delete_block(user, blocked) do
@cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
@cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
{:ok, relationship}
end
end

View file

@ -4,6 +4,7 @@
defmodule Pleroma.User.Search do
alias Pleroma.EctoType.ActivityPub.ObjectValidators.Uri, as: UriType
alias Pleroma.Instances.Instance
alias Pleroma.Pagination
alias Pleroma.User
@ -88,12 +89,13 @@ defmodule Pleroma.User.Search do
|> filter_invisible_users()
|> filter_internal_users()
|> filter_blocked_domains(for_user)
|> filter_unreachable_users()
|> fts_search(query_string)
|> select_top_users(top_user_ids)
|> trigram_rank(query_string)
|> boost_search_rank(for_user, top_user_ids)
|> subquery()
|> order_by(desc: :search_rank)
|> order_by_search_rank(for_user)
|> maybe_restrict_local(for_user)
|> maybe_restrict_accepting_chat_messages(capabilities)
|> filter_deactivated_users()
@ -196,6 +198,14 @@ defmodule Pleroma.User.Search do
defp filter_blocked_domains(query, _), do: query
defp filter_unreachable_users(query) do
from(u in query,
left_join: i in Instance,
on: i.host == fragment("substring(? from '.*://([^/]*)')", u.ap_id),
where: is_nil(i.unreachable_since)
)
end
defp maybe_resolve(true, user, query) do
case {limit(), user} do
{:all, _} -> :noop
@ -236,6 +246,16 @@ defmodule Pleroma.User.Search do
from(u in subquery(query),
select_merge: %{
search_type:
fragment(
"""
CASE WHEN (?) THEN 2
WHEN (?) THEN 1
ELSE 0 END
""",
u.id in ^top_user_ids,
u.id in ^friends_ids or u.id in ^followers_ids
),
search_rank:
fragment(
"""
@ -261,6 +281,14 @@ defmodule Pleroma.User.Search do
defp boost_search_rank(query, _for_user, top_user_ids) do
from(u in subquery(query),
select_merge: %{
search_type:
fragment(
"""
CASE WHEN (?) THEN 2
ELSE 0 END
""",
u.id in ^top_user_ids
),
search_rank:
fragment(
"""
@ -273,4 +301,22 @@ defmodule Pleroma.User.Search do
}
)
end
defp order_by_search_rank(query, %User{}) do
order_by(
query,
[u],
desc: u.search_type,
desc_nulls_last:
fragment(
"CASE WHEN ? = 1 THEN COALESCE(?, ?) ELSE NULL END",
u.search_type,
u.last_status_at,
u.last_active_at
),
desc: u.search_rank
)
end
defp order_by_search_rank(query, _), do: order_by(query, desc: :search_rank)
end

View file

@ -30,7 +30,7 @@ defmodule Pleroma.Web do
def controller do
quote do
use Phoenix.Controller, namespace: Pleroma.Web
use Phoenix.Controller, formats: [json: "View", html: "View"]
import Plug.Conn
@ -42,7 +42,10 @@ defmodule Pleroma.Web do
plug(:set_put_layout)
defp set_put_layout(conn, _) do
put_layout(conn, Pleroma.Config.get(:app_layout, "app.html"))
case Pleroma.Config.get(:app_layout, "app.html") do
false -> put_layout(conn, false)
layout -> put_layout(conn, {Pleroma.Web.LayoutView, layout})
end
end
# Marks plugs intentionally skipped and blocks their execution if present in plugs chain

View file

@ -537,9 +537,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
# and extra sorting on "activities.id DESC NULLS LAST" would worse the query plan
opts = Map.put(opts, :skip_extra_order, true)
Pagination.fetch_paginated(query, opts, pagination)
Pagination.fetch_paginated(query, opts, pagination, pagination_binding(opts))
end
defp pagination_binding(%{favorited_by: _}), do: :favorited_activity
defp pagination_binding(_), do: nil
def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
list_memberships = Pleroma.List.memberships(opts[:user])
@ -786,6 +789,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
defp restrict_since(query, %{since_id: ""}), do: query
defp restrict_since(query, %{favorited_by: _, since_id: _}), do: query
defp restrict_since(query, %{since_id: since_id}) do
from(activity in query, where: activity.id > ^since_id)
end
@ -893,14 +898,22 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
|> Repo.all()
# Note: NO extra ordering should be done on "activities.id desc nulls last" for optimal plan
from(
[_activity, object] in query,
join: hto in "hashtags_objects",
on: hto.object_id == object.id,
where: hto.hashtag_id in ^hashtag_ids,
distinct: [desc: object.id],
order_by: [desc: object.id]
)
query =
from(
[_activity, object] in query,
join: hto in "hashtags_objects",
on: hto.object_id == object.id,
where: hto.hashtag_id in ^hashtag_ids
)
if is_nil(query.distinct) do
from([_activity, object] in query,
distinct: [desc: object.id],
order_by: [desc: object.id]
)
else
query
end
end
defp restrict_hashtag_any(query, %{tag: tag}) when is_binary(tag) do
@ -1012,9 +1025,31 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
defp restrict_assigned_account(query, _), do: query
defp restrict_favorited_by(query, %{favorited_by: ap_id}) do
newer_favorite =
from(newer_favorite in Activity,
where: newer_favorite.actor == ^ap_id,
where: newer_favorite.id > parent_as(:favorited_activity).id,
where: fragment("?->>'type' = ?", newer_favorite.data, "Like"),
where:
fragment(
"associated_object_id(?) = associated_object_id(?)",
newer_favorite.data,
parent_as(:favorited_activity).data
),
select: 1
)
from(
[_activity, object] in query,
where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
[activity, object: object] in query,
join: favorited_activity in Activity,
as: :favorited_activity,
on:
favorited_activity.actor == ^ap_id and
fragment("?->>'type' = ?", favorited_activity.data, "Like") and
fragment("associated_object_id(?) = (?)->>'id'", favorited_activity.data, object.data),
where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id),
where: not exists(subquery(newer_favorite)),
select: %Activity{activity | pagination_id: favorited_activity.id}
)
end
@ -1881,11 +1916,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
end
end
@featured_collection_types ["OrderedCollection", "Collection"]
@featured_collection_page_types ["OrderedCollectionPage", "CollectionPage"]
@featured_collection_item_types @featured_collection_types ++ @featured_collection_page_types
def pin_data_from_featured_collection(%{
"type" => type,
"orderedItems" => objects
})
when type in ["OrderedCollection", "Collection"] do
when type in @featured_collection_item_types do
Map.new(objects, fn
%{"id" => object_ap_id} -> {object_ap_id, NaiveDateTime.utc_now()}
object_ap_id when is_binary(object_ap_id) -> {object_ap_id, NaiveDateTime.utc_now()}
@ -1903,7 +1942,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
def fetch_and_prepare_featured_from_ap_id(ap_id) do
with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
{:ok, pin_data_from_featured_collection(data)}
{:ok, prepare_featured_collection(data)}
else
e ->
Logger.error("Could not decode featured collection at fetch #{ap_id}, #{inspect(e)}")
@ -1911,6 +1950,34 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
end
end
defp prepare_featured_collection(%{"orderedItems" => objects} = data) when is_list(objects) do
pin_data_from_featured_collection(data)
end
defp prepare_featured_collection(%{
"type" => type,
"first" => %{"type" => page_type} = first
})
when type in @featured_collection_types and page_type in @featured_collection_page_types do
pin_data_from_featured_collection(first)
end
defp prepare_featured_collection(%{"type" => type, "first" => first})
when type in @featured_collection_types and is_binary(first) do
case Fetcher.fetch_and_contain_remote_object_from_id(first) do
{:ok, data} ->
pin_data_from_featured_collection(data)
e ->
Logger.error("Could not decode featured collection page at fetch #{first}, #{inspect(e)}")
%{}
end
end
defp prepare_featured_collection(data) do
pin_data_from_featured_collection(data)
end
def enqueue_pin_fetches(%{pinned_objects: pins}) do
# enqueue a task to fetch all pinned objects
Enum.each(pins, fn {ap_id, _} ->

View file

@ -303,7 +303,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
end
end
def inbox(%{assigns: %{valid_signature: true}} = conn, %{"nickname" => nickname} = params) do
def inbox(
%{
assigns: %{valid_signature: true, valid_host_header: true}
} = conn,
%{"nickname" => nickname} = params
) do
with {:recipient_exists, %User{} = recipient} <-
{:recipient_exists, User.get_cached_by_nickname(nickname)},
{:sender_exists, {:ok, %User{} = actor}} <-
@ -342,7 +347,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
end
end
def inbox(%{assigns: %{valid_signature: true}} = conn, params) do
def inbox(%{assigns: %{valid_signature: true, valid_host_header: true}} = conn, params) do
Federator.incoming_ap_doc(params)
json(conn, "ok")
end

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.ActivityExpirationPolicy do
@moduledoc "Adds expiration to all local Create activities"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def filter(activity) do
activity =

View file

@ -3,11 +3,11 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.AntiFollowbotPolicy do
@moduledoc "Prevent followbots from following with a bit of heuristic"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.User
@moduledoc "Prevent followbots from following with a bit of heuristic"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
# XXX: this should become User.normalize_by_ap_id() or similar, really.
defp normalize_by_ap_id(%{"id" => id}), do: User.get_cached_by_ap_id(id)

View file

@ -3,9 +3,11 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.AntiLinkSpamPolicy do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.User
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
require Logger

View file

@ -3,11 +3,13 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.AntiMentionSpamPolicy do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.Config
alias Pleroma.User
require Pleroma.Constants
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp user_has_posted?(%User{} = u), do: u.note_count > 0

View file

@ -3,10 +3,12 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.DropPolicy do
require Logger
@moduledoc "Drop and log everything received"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Logger
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def filter(activity) do
Logger.debug("REJECTING #{inspect(activity)}")

View file

@ -3,6 +3,8 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.EmojiPolicy do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants
alias Pleroma.Object.Updater
@ -10,7 +12,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.EmojiPolicy do
@moduledoc "Reject or force-unlisted emojis with certain URLs or names"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp config_remove_url do
Pleroma.Config.get([:mrf_emoji, :remove_url], [])

View file

@ -3,11 +3,13 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.EnsureRePrepended do
alias Pleroma.Object
@moduledoc "Ensure a re: is prepended on replies to a post with a Subject"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.Object
use Pleroma.Web.ActivityPub.MRF.Policy
@reply_prefix Regex.compile!("^re:[[:space:]]*", [:caseless])
def history_awareness, do: :auto

View file

@ -4,12 +4,15 @@
defmodule Pleroma.Web.ActivityPub.MRF.FollowBotPolicy do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.Config
alias Pleroma.User
alias Pleroma.Web.CommonAPI
require Logger
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def filter(activity) do
with follower_nickname <- Config.get([:mrf_follow_bot, :follower_nickname]),

View file

@ -3,9 +3,12 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.ForceBotUnlistedPolicy do
alias Pleroma.User
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
@moduledoc "Remove bot posts from federated timeline"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.User
use Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants

View file

@ -3,13 +3,14 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.ForceMention do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants
alias Pleroma.Config
alias Pleroma.Object
alias Pleroma.User
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp get_author(url) do
with %Object{data: %{"actor" => actor}} <- Object.normalize(url, fetch: false),

View file

@ -3,13 +3,15 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.ForceMentionsInContent do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants
alias Pleroma.Formatter
alias Pleroma.Object
alias Pleroma.User
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def history_awareness, do: :auto

View file

@ -3,6 +3,8 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.HashtagPolicy do
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants
alias Pleroma.Config
@ -14,7 +16,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.HashtagPolicy do
Note: This MRF Policy is always enabled, if you want to disable it you have to set empty lists.
"""
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def history_awareness, do: :manual

View file

@ -3,13 +3,14 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.HellthreadPolicy do
@moduledoc "Block activities with too much mentions (configurable)"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.User
require Pleroma.Constants
@moduledoc "Block activities with too much mentions (configurable)"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp delist_activity(activity, threshold) when threshold > 0 do
follower_collection = User.get_cached_by_ap_id(activity["actor"]).follower_address

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.InlineQuotePolicy do
@moduledoc "Force a quote line into the message content."
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp build_inline_quote(template, url) do
quote_line = String.replace(template, "{url}", "<a href=\"#{url}\">#{url}</a>")

View file

@ -3,13 +3,14 @@
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.ActivityPub.MRF.KeywordPolicy do
@moduledoc "Reject or Word-Replace activities with a keyword or regex"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
require Pleroma.Constants
alias Pleroma.Web.ActivityPub.MRF.Utils
@moduledoc "Reject or Word-Replace activities with a keyword or regex"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
defp string_matches?(string, pattern) when is_binary(pattern) do
String.contains?(string, pattern)

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
@moduledoc "Preloads any attachments in the MediaProxy cache by prefetching them"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.HTTP
alias Pleroma.Web.MediaProxy

View file

@ -4,9 +4,10 @@
defmodule Pleroma.Web.ActivityPub.MRF.MentionPolicy do
@moduledoc "Block activities which mention a user"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def filter(%{"type" => "Create"} = activity) do
reject_actors = Pleroma.Config.get([:mrf_mention, :actors], [])

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoEmptyPolicy do
@moduledoc "Filter local activities which have no content"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
alias Pleroma.Web.Endpoint
@impl true

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoOpPolicy do
@moduledoc "Does nothing (lets the messages go through unmodified)"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def filter(activity) do
{:ok, activity}

View file

@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoPlaceholderTextPolicy do
@moduledoc "Ensure no content placeholder is present (such as the dot from mastodon)"
@behaviour Pleroma.Web.ActivityPub.MRF.Policy
use Pleroma.Web.ActivityPub.MRF.Policy
@impl true
def history_awareness, do: :auto

Some files were not shown because too many files have changed in this diff Show more