Merge pull request #7583 from pixelfed/staging

Add Block Sync + Incoming & Outgoing Block Activities (FEP-070c)
pull/7591/head
dansup 4 weeks ago committed by GitHub
commit 05c01012de
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -0,0 +1,85 @@
<?php
namespace App\Console\Commands\Internal;
use App\Jobs\Federation\BlockSyncPipeline;
use App\Services\BlockSyncService;
use App\Services\DeliveryHostService;
use App\Services\FollowersSyncService;
use App\Services\InstanceService;
use Illuminate\Console\Attributes\Description;
use Illuminate\Console\Attributes\Signature;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
#[Signature('federation:block-sync-reconcile {--limit= : Maximum peers to reconcile} {--spread=3600 : Seconds to spread the jobs over}')]
#[Description('FEP-070c: reconcile blocks received from peers with their synchronization endpoints')]
class BlockSyncReconcile extends Command
{
const CURSOR_KEY = 'pf:services:block-sync:reconcile-cursor';
public function handle(): int
{
if (! BlockSyncService::receiving()) {
$this->info('Block synchronization is not enabled.');
return self::SUCCESS;
}
$limit = max(1, (int) ($this->option('limit') ?? config('federation.activitypub.block_sync.reconcile_limit', 1000)));
$spread = max(0, (int) $this->option('spread'));
$banned = (array) InstanceService::getBannedDomains();
$cursor = (int) Cache::get(self::CURSOR_KEY, 0);
$lastId = $cursor;
$dispatched = 0;
$peers = fn () => DB::table('instances')
->whereNotNull('block_sync_url')
->where('banned', false)
->select('id', 'domain', 'block_sync_url');
$process = function ($instances) use (&$dispatched, &$lastId, $limit, $spread, $banned) {
foreach ($instances as $instance) {
if ($dispatched >= $limit) {
return false;
}
$lastId = (int) $instance->id;
$domain = strtolower((string) $instance->domain);
$url = (string) $instance->block_sync_url;
$authority = FollowersSyncService::authority($url);
if (
! $authority
|| strtolower((string) parse_url($authority, PHP_URL_HOST)) !== $domain
|| in_array($domain, $banned, true)
|| DeliveryHostService::isUnavailable($domain)
) {
continue;
}
BlockSyncPipeline::dispatch($authority, $url, true)
->onQueue('low')
->delay(now()->addSeconds($spread > 0 ? random_int(0, $spread) : 0));
$dispatched++;
}
return true;
};
$peers()->where('id', '>', $cursor)->chunkById(500, $process);
if ($dispatched < $limit && $cursor > 0) {
$peers()->where('id', '<=', $cursor)->chunkById(500, $process);
}
Cache::forever(self::CURSOR_KEY, $dispatched >= $limit ? $lastId : 0);
$this->info("Queued {$dispatched} block synchronizations.");
return self::SUCCESS;
}
}

@ -92,6 +92,7 @@ These are primarily invoked by the scheduler (see `bootstrap/app.php`) rather th
| ------------------------------------------ | ------------------------------------------------------- |
| `app:account-post-count-stat-update` | Update post counts from recent activity. |
| `app:cleanup-expired-app-registrations` | Delete app registrations older than 90 days. |
| `federation:block-sync-reconcile` | FEP-070c: reconcile blocks received from peers. |
| `gc:sessions` | Garbage-collect database sessions. |
| `gc:failedjobs` | Delete failed jobs older than one month. |
| `app:hashtag-cached-count-update` | Update cached hashtag counters (`--limit`). |

@ -0,0 +1,52 @@
<?php
namespace App\Federation\ActivityBuilders;
use App\Models\Profile;
class BlockActivityBuilder
{
/**
* Block of a remote actor by a local actor.
*
* @return array<string, mixed>
*/
public function build(Profile $blocker, Profile $blocked, int $filterId): array
{
return [
...$this->object($blocker, $blocked, $filterId),
'@context' => 'https://www.w3.org/ns/activitystreams',
];
}
/**
* @return array<string, mixed>
*/
public function buildUndo(Profile $blocker, Profile $blocked, int $filterId): array
{
$actorId = $blocker->permalink();
return [
'@context' => 'https://www.w3.org/ns/activitystreams',
'id' => $actorId.'#blocks/'.$filterId.'/undo',
'type' => 'Undo',
'actor' => $actorId,
'object' => $this->object($blocker, $blocked, $filterId),
];
}
/**
* @return array<string, mixed>
*/
protected function object(Profile $blocker, Profile $blocked, int $filterId): array
{
$actorId = $blocker->permalink();
return [
'id' => $actorId.'#blocks/'.$filterId,
'type' => 'Block',
'actor' => $actorId,
'object' => $blocked->remote_url,
];
}
}

@ -12,6 +12,7 @@ use App\Models\QuoteAuthorization;
use App\Models\Status;
use App\Services\AccountService;
use App\Services\ActivityPubSignedFetchService;
use App\Services\BlockSyncService;
use App\Services\FeaturedCollectionService;
use App\Services\FollowersSyncService;
use App\Services\InstanceService;
@ -365,6 +366,56 @@ class FederationController extends Controller
->header('Cache-Control', 'private, no-store');
}
public function blockSynchronization(Request $request): Response|JsonResponse
{
abort_if(! BlockSyncService::disclosing(), 404);
$signer = ActivityPubSignedFetchService::verify($request);
abort_if(! $signer, 401);
$authority = FollowersSyncService::authority($signer->remote_url);
abort_if(! $authority, 401);
abort_if(! BlockSyncService::isDisclosedPeer($authority), 403);
$pairs = BlockSyncService::disclosedPairs($authority);
$digest = BlockSyncService::digest($pairs);
$etag = '"'.$digest.'"';
$headers = [
'Cache-Control' => 'private, no-store',
'ETag' => $etag,
];
$ifNoneMatch = array_map(
fn (string $tag) => preg_replace('#^W/#', '', trim($tag)),
explode(',', (string) $request->headers->get('If-None-Match', ''))
);
if (in_array($etag, $ifNoneMatch, true)) {
return response('', 304, $headers);
}
$res = [
'@context' => [
'https://www.w3.org/ns/activitystreams',
[
'blockSynchronizationDigest' => BlockSyncService::DIGEST_CONTEXT_TERM,
],
],
'id' => BlockSyncService::endpointUrl(),
'type' => 'OrderedCollection',
'totalItems' => count($pairs),
'blockSynchronizationDigest' => $digest,
'orderedItems' => array_map(fn (array $pair) => [
'type' => 'Block',
'actor' => $pair['actor'],
'object' => $pair['object'],
], $pairs),
];
return response()
->json($res, 200, $headers, JSON_UNESCAPED_SLASHES)
->header('Content-Type', 'application/activity+json');
}
public function userFeatureAuthorization(Request $request, $username, $id): JsonResponse
{
abort_if(! (bool) config_cache('federation.activitypub.enabled'), 404);

@ -3,6 +3,7 @@
namespace App\Http\Controllers;
use App\Models\InstanceActor;
use App\Services\BlockSyncService;
use Illuminate\Http\Response;
use Illuminate\Support\Facades\Cache;
@ -16,6 +17,14 @@ class InstanceActorController extends Controller
return json_encode($res, JSON_UNESCAPED_SLASHES);
});
if (BlockSyncService::disclosing()) {
$actor = json_decode((string) $res, true);
if (is_array($actor)) {
$res = json_encode(BlockSyncService::decorateInstanceActor($actor), JSON_UNESCAPED_SLASHES) ?: $res;
}
}
return response($res)->header('Content-Type', 'application/activity+json');
}

@ -0,0 +1,90 @@
<?php
namespace App\Jobs\Federation;
use App\Services\BlockSyncService;
use App\Services\DeliveryHostService;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\Middleware\WithoutOverlapping;
use Illuminate\Support\Facades\Log;
use Throwable;
class BlockSyncPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable;
public $timeout = 120;
public $tries = 1;
public const REMOTE_BUDGET = 90;
public function __construct(
protected string $authority,
protected string $url,
protected bool $periodic = false
) {}
public function uniqueId(): string
{
return ($this->periodic ? 'pull:' : 'push:').hash('sha256', $this->authority);
}
public function uniqueFor(): int
{
return $this->periodic
? 5400
: max(60, (int) config('federation.activitypub.block_sync.cooldown', 900));
}
/**
* @return array<int, object>
*/
public function middleware(): array
{
return [
(new WithoutOverlapping('block-sync:'.hash('sha256', $this->authority)))
->shared()
->dontRelease()
->expireAfter($this->timeout + 30),
];
}
public function handle(): void
{
$domain = DeliveryHostService::domain($this->authority);
if ($domain && DeliveryHostService::isUnavailable($domain)) {
return;
}
$signed = BlockSyncService::pullSignedDigest($this->authority);
try {
$result = BlockSyncService::synchronize(
$this->authority,
$signed['url'] ?? $this->url,
$signed['digest'] ?? null,
microtime(true) + self::REMOTE_BUDGET
);
} catch (Throwable $e) {
Log::warning('BlockSync: synchronization failed', [
'authority' => $this->authority,
'exception' => $e::class,
'error' => $e->getMessage(),
]);
BlockSyncService::backOff($this->authority);
return;
}
if ($result['status'] === 'fetch_failed') {
BlockSyncService::backOff($this->authority);
}
}
}

@ -0,0 +1,118 @@
<?php
namespace App\Jobs\Federation;
use App\Federation\ActivityBuilders\BlockActivityBuilder;
use App\Models\Profile;
use App\Models\UserFilter;
use App\Services\ActivityPubDeliveryService;
use App\Services\BlockSyncService;
use DateTimeInterface;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\Middleware\WithoutOverlapping;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;
use Throwable;
class DeliverBlockActivity implements ShouldBeUniqueUntilProcessing, ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public const RETRY_DELAYS = [300, 3600];
public $timeout = 60;
public $uniqueFor = 7200;
public function __construct(
protected int $blockerId,
protected int $blockedId,
protected int $filterId,
protected bool $undo = false,
protected int $pass = 1
) {}
public function uniqueId(): string
{
return $this->blockerId.':'.$this->blockedId;
}
public function middleware(): array
{
return [
(new WithoutOverlapping('block-delivery:'.$this->uniqueId()))
->releaseAfter(15)
->expireAfter(120),
];
}
public function retryUntil(): DateTimeInterface
{
return now()->addMinutes(10);
}
public function handle(): void
{
if (! BlockSyncService::disclosing()) {
return;
}
$blocker = Profile::whereNull('domain')
->whereNull('status')
->find($this->blockerId);
$blocked = Profile::whereNotNull('domain')->find($this->blockedId);
if (! $blocker || empty($blocker->private_key) || ! $blocked || ! $blocked->inbox_url) {
return;
}
$filter = UserFilter::whereUserId($this->blockerId)
->whereFilterableId($this->blockedId)
->whereFilterableType(Profile::class)
->whereFilterType('block')
->first(['id']);
$blocking = $filter !== null;
if ($blocking === $this->undo) {
return;
}
$builder = new BlockActivityBuilder;
$activity = $this->undo
? $builder->buildUndo($blocker, $blocked, $this->filterId)
: $builder->build($blocker, $blocked, (int) $filter->id);
try {
$result = ActivityPubDeliveryService::pool($blocker, [$blocked->inbox_url], $activity);
} catch (Throwable $e) {
Log::debug('DeliverBlockActivity: delivery failed', [
'blocker_id' => $this->blockerId,
'blocked_id' => $this->blockedId,
'error' => $e->getMessage(),
]);
return;
}
if ($result['failed'] === 0 || $this->pass > count(self::RETRY_DELAYS)) {
return;
}
self::dispatch(
$this->blockerId,
$this->blockedId,
$this->undo ? $this->filterId : (int) $filter->id,
$this->undo,
$this->pass + 1
)
->onQueue('high')
->delay(now()->addSeconds(self::RETRY_DELAYS[$this->pass - 1]));
}
}

@ -4,6 +4,7 @@ namespace App\Jobs\InboxPipeline;
use App\Jobs\InboxPipeline\Concerns\RetriesWhenActorUnavailable;
use App\Models\Profile;
use App\Services\BlockSyncService;
use App\Services\FollowersSyncService;
use App\Util\ActivityPub\Helpers;
use App\Util\ActivityPub\HttpSignature;
@ -84,6 +85,9 @@ class InboxValidator implements ShouldQueue
// FEP-8fcf: compare the sender's followers digest with our copy
FollowersSyncService::handleInboundHeaders($headers);
// FEP-070c: compare the sender's block digest with our copy
BlockSyncService::handleInboundHeaders($headers);
if (isset($payload['type']) && in_array($payload['type'], ['Follow', 'Accept'])) {
ActivityHandler::dispatch($headers, $profile, $payload)->onQueue('follow');
} else {

@ -4,6 +4,7 @@ namespace App\Jobs\InboxPipeline;
use App\Jobs\InboxPipeline\Concerns\RetriesWhenActorUnavailable;
use App\Models\Profile;
use App\Services\BlockSyncService;
use App\Services\FollowersSyncService;
use App\Util\ActivityPub\Helpers;
use App\Util\ActivityPub\HttpSignature;
@ -70,6 +71,9 @@ class InboxWorker implements ShouldQueue
// FEP-8fcf: compare the sender's followers digest with our copy
FollowersSyncService::handleInboundHeaders($headers);
// FEP-070c: compare the sender's block digest with our copy
BlockSyncService::handleInboundHeaders($headers);
ActivityHandler::dispatch($headers, $profile, $payload)->onQueue('shared');
return;

@ -37,6 +37,7 @@ use Illuminate\Support\Carbon;
* @property string|null $shared_inbox
* @property int|null $allowlisted
* @property int $delivery_failures
* @property string|null $block_sync_url
* @property-read Collection<int, Media> $media
* @property-read int|null $media_count
* @property-read Collection<int, Profile> $profiles

@ -6,6 +6,7 @@ use App\Jobs\HomeFeedPipeline\FeedFollowPipeline;
use App\Jobs\HomeFeedPipeline\FeedUnfollowPipeline;
use App\Models\Profile;
use App\Models\UserFilter;
use App\Services\BlockSyncService;
use App\Services\FeaturedCollectionService;
use App\Services\QuoteService;
use App\Services\UserFilterService;
@ -36,7 +37,9 @@ class UserFilterObserver
*/
public function updated(UserFilter $userFilter)
{
$this->filterCreate($userFilter);
// Not a new block: re-running the federation side would toggle the
// pair out of the FEP-070c digest and send a duplicate Block.
$this->filterCreate($userFilter, false);
}
/**
@ -69,7 +72,7 @@ class UserFilterObserver
$this->filterDelete($userFilter);
}
protected function filterCreate(UserFilter $userFilter)
protected function filterCreate(UserFilter $userFilter, bool $federate = true)
{
if ($userFilter->filterable_type !== Profile::class) {
return;
@ -83,10 +86,19 @@ class UserFilterObserver
case 'block':
UserFilterService::block($userFilter->user_id, $userFilter->filterable_id);
if ($this->isRemoteBlocker($userFilter)) {
FeedUnfollowPipeline::dispatch($userFilter->filterable_id, $userFilter->user_id)->onQueue('feed');
break;
}
FeedUnfollowPipeline::dispatch($userFilter->user_id, $userFilter->filterable_id)->onQueue('feed');
// user_id is the blocking profile id, filterable_id the blocked profile
FeaturedCollectionService::revokeForActor($userFilter->user_id, $userFilter->filterable_id);
QuoteService::revokeForActor($userFilter->user_id, $userFilter->filterable_id);
if ($federate) {
BlockSyncService::localBlockChanged($userFilter, true);
}
break;
}
}
@ -105,8 +117,22 @@ class UserFilterObserver
case 'block':
UserFilterService::unblock($userFilter->user_id, $userFilter->filterable_id);
if ($this->isRemoteBlocker($userFilter)) {
break;
}
FeedFollowPipeline::dispatch($userFilter->user_id, $userFilter->filterable_id)->onQueue('feed');
BlockSyncService::localBlockChanged($userFilter, false);
break;
}
}
protected function isRemoteBlocker(UserFilter $userFilter): bool
{
return Profile::withTrashed()
->whereKey($userFilter->user_id)
->whereNotNull('domain')
->exists();
}
}

@ -276,6 +276,19 @@ class ActivityPubDeliveryService
}
}
$blockSyncDigests = null;
if (BlockSyncService::disclosing()) {
try {
$blockSyncDigests = BlockSyncService::outboundDigests();
} catch (Throwable $e) {
Log::warning('Unable to compute block synchronization digests', [
'exception' => $e::class,
'error' => $e->getMessage(),
]);
}
}
/*
* Prepare and sign every delivery before starting the HTTP pool.
*
@ -347,6 +360,14 @@ class ActivityPubDeliveryService
}
}
if ($blockSyncDigests !== null) {
$blockSyncHeader = BlockSyncService::header($url, $blockSyncDigests);
if ($blockSyncHeader !== null) {
$extraHeaders[BlockSyncService::HEADER] = $blockSyncHeader;
}
}
$headers = self::signedHeaders(
$profile,
$url,

File diff suppressed because it is too large Load Diff

@ -3,6 +3,7 @@
namespace App\Util\ActivityPub;
use App\Util\ActivityPub\Inbox\HandlesAnnouncements;
use App\Util\ActivityPub\Inbox\HandlesBlocks;
use App\Util\ActivityPub\Inbox\HandlesCreates;
use App\Util\ActivityPub\Inbox\HandlesDeletes;
use App\Util\ActivityPub\Inbox\HandlesFeatureRequests;
@ -28,6 +29,7 @@ use Illuminate\Support\Facades\Log;
class Inbox
{
use HandlesAnnouncements;
use HandlesBlocks;
use HandlesCreates;
use HandlesDeletes;
use HandlesFeatureRequests;
@ -70,6 +72,10 @@ class Inbox
$this->handleAddActivity();
break;
case 'Block':
$this->handleBlockActivity();
break;
case 'Create':
$this->handleCreateActivity();
break;

@ -0,0 +1,79 @@
<?php
namespace App\Util\ActivityPub\Inbox;
use App\Models\Profile;
use App\Services\BlockSyncService;
use App\Services\FollowersSyncService;
trait HandlesBlocks
{
public function handleBlockActivity(): void
{
if (! BlockSyncService::receiving()) {
return;
}
$actorUrl = BlockSyncService::idOf($this->payload['actor'] ?? null);
$objectUrl = BlockSyncService::idOf($this->payload['object'] ?? null);
if (! $actorUrl || ! $objectUrl) {
return;
}
$target = FollowersSyncService::resolveLocalActor($objectUrl);
if (! $target) {
return;
}
$actor = $this->validateAndFetchActor($actorUrl);
if (! $actor || $actor->domain === null || ! $this->signedFromActorAuthority($actor)) {
return;
}
BlockSyncService::applyRemoteBlock($actor, $target);
}
protected function handleUndoBlock(Profile $profile, array $obj): void
{
if (! BlockSyncService::receiving() || $profile->domain === null) {
return;
}
$blockActor = BlockSyncService::idOf($obj['actor'] ?? null);
if ($blockActor !== null && $blockActor !== $profile->remote_url) {
return;
}
$objectUrl = BlockSyncService::idOf($obj['object'] ?? null);
if (! $objectUrl) {
return;
}
$target = FollowersSyncService::resolveLocalActor($objectUrl);
if (! $target || ! $this->signedFromActorAuthority($profile)) {
return;
}
BlockSyncService::removeRemoteBlock($profile, $target);
}
protected function signedFromActorAuthority(Profile $actor): bool
{
$signer = $this->signingProfile();
if (! $signer || $signer->domain === null) {
return false;
}
$authority = FollowersSyncService::authority($actor->remote_url);
return $authority !== null
&& FollowersSyncService::authority($signer->remote_url) === $authority;
}
}

@ -34,7 +34,7 @@ trait HandlesUndos
match ($obj['type']) {
'Accept' => null,
'Announce' => $this->handleUndoAnnounce($profile, $obj),
'Block' => null,
'Block' => $this->handleUndoBlock($profile, $obj),
'Follow' => $this->handleUndoFollow($profile, $obj),
'Like' => $this->handleUndoLike($profile, $obj),
default => null,

@ -51,3 +51,7 @@ if ((bool) config('scheduledtasks.account_storage_reconcile')) {
}
$schedule->command('app:instance-update-total-local-posts')->twiceDailyAt(1, 13, 45)->onOneServer();
if ((bool) config('federation.activitypub.block_sync.enabled', true)) {
$schedule->command('federation:block-sync-reconcile')->dailyAt('03:40')->onOneServer()->withoutOverlapping(120);
}

@ -52,6 +52,14 @@ return [
// Maximum collection pages fetched from a remote server per synchronization
'max_pages' => env('AP_FOLLOWERS_SYNC_MAX_PAGES', 10),
],
'block_sync' => [
'enabled' => env('AP_BLOCK_SYNC', true),
'disclose' => env('AP_BLOCK_SYNC_DISCLOSE', true),
'cooldown' => env('AP_BLOCK_SYNC_COOLDOWN', 900),
'max_pages' => env('AP_BLOCK_SYNC_MAX_PAGES', 10),
'reconcile_limit' => env('AP_BLOCK_SYNC_RECONCILE_LIMIT', 1000),
],
],
'atom' => [

@ -0,0 +1,26 @@
<?php
use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;
return new class extends Migration
{
public function up(): void
{
Schema::table('instances', function (Blueprint $table) {
if (! Schema::hasColumn('instances', 'block_sync_url')) {
$table->string('block_sync_url')->nullable();
}
});
}
public function down(): void
{
Schema::table('instances', function (Blueprint $table) {
if (Schema::hasColumn('instances', 'block_sync_url')) {
$table->dropColumn('block_sync_url');
}
});
}
};

@ -47,6 +47,7 @@ $middleware = ['auth:sanctum,api'];
Route::post('/f/inbox', [FederationController::class, 'sharedInbox']);
Route::post('/users/{username}/inbox', [FederationController::class, 'userInbox']);
Route::get('/users/{username}/followers_synchronization', [FederationController::class, 'userFollowersSynchronization']);
Route::get('/f/block_sync', [FederationController::class, 'blockSynchronization']);
Route::get('i/actor', [InstanceActorController::class, 'profile']);
Route::post('i/actor/inbox', [InstanceActorController::class, 'inbox']);
Route::get('i/actor/outbox', [InstanceActorController::class, 'outbox']);

@ -0,0 +1,575 @@
<?php
use App\Jobs\Federation\BlockSyncPipeline;
use App\Jobs\Federation\DeliverBlockActivity;
use App\Models\Follower;
use App\Models\Instance;
use App\Models\InstanceActor;
use App\Models\Profile;
use App\Models\User;
use App\Models\UserFilter;
use App\Services\ActivityPubDeliveryService;
use App\Services\BlockSyncService;
use App\Util\ActivityPub\Inbox;
use Illuminate\Foundation\Testing\LazilyRefreshDatabase;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Http;
use Illuminate\Support\Facades\Queue;
use Illuminate\Support\Facades\Redis;
uses(LazilyRefreshDatabase::class);
beforeEach(function () {
Redis::spy();
Queue::fake();
config([
'instance.enable_cc' => false,
'federation.activitypub.enabled' => true,
'federation.activitypub.block_sync.enabled' => true,
'federation.activitypub.block_sync.disclose' => true,
]);
});
function bsyncLocalProfile(): Profile
{
$user = User::factory()->create();
$user->refresh();
return $user->profile;
}
function bsyncRemoteProfile(string $domain, string $username, array $attributes = []): Profile
{
$actor = "https://{$domain}/users/{$username}";
return Profile::factory()->remote()->create(array_merge([
'domain' => $domain,
'username' => "@{$username}@{$domain}",
'remote_url' => $actor,
'inbox_url' => "{$actor}/inbox",
'sharedInbox' => "https://{$domain}/inbox",
'key_id' => "{$actor}#main-key",
'last_fetched_at' => now(),
], $attributes));
}
function bsyncBlock(Profile $blocker, Profile $blocked, int $ageInMinutes = 120): void
{
DB::table('user_filters')->insert([
'user_id' => $blocker->id,
'filterable_id' => $blocked->id,
'filterable_type' => Profile::class,
'filter_type' => 'block',
'created_at' => now()->subMinutes($ageInMinutes),
'updated_at' => now()->subMinutes($ageInMinutes),
]);
}
function bsyncBlocks(Profile $blocker, Profile $blocked): bool
{
return UserFilter::whereUserId($blocker->id)
->whereFilterableId($blocked->id)
->whereFilterableType(Profile::class)
->whereFilterType('block')
->exists();
}
function bsyncPair(Profile $blocker, Profile $blocked): array
{
return ['actor' => $blocker->permalink(), 'object' => $blocked->permalink()];
}
function bsyncSeedHosts(array $hosts): void
{
foreach ($hosts as $host) {
Cache::put('helpers:url:public-ips:'.hash('xxh128', $host), ['203.0.113.40'], 3600);
}
Cache::put('instances:banned:domains', [], 1209600);
Cache::put('instances:unlisted:domains', [], 1209600);
}
function bsyncKeyPair(): array
{
$key = openssl_pkey_new([
'private_key_bits' => 2048,
'private_key_type' => OPENSSL_KEYTYPE_RSA,
]);
openssl_pkey_export($key, $private);
return [$private, openssl_pkey_get_details($key)['key']];
}
function bsyncInProduction(callable $fn): mixed
{
$app = app();
$previous = $app['env'];
$app['env'] = 'production';
try {
return $fn();
} finally {
$app['env'] = $previous;
}
}
function bsyncSignedGetHeaders(string $privateKey, string $keyId, string $path, string $host = 'pixelfed.test'): array
{
$date = now()->toRfc7231String();
openssl_sign(
"(request-target): get {$path}\nhost: {$host}\ndate: {$date}",
$signature,
$privateKey,
OPENSSL_ALGO_SHA256
);
return [
'Accept' => 'application/activity+json',
'Date' => $date,
'Signature' => sprintf(
'keyId="%s",algorithm="rsa-sha256",headers="(request-target) host date",signature="%s"',
$keyId,
base64_encode($signature)
),
];
}
function bsyncInboundHeaders(Profile $sender, string $url, string $digest, bool $signed = true): array
{
$list = '(request-target) host date digest'.($signed ? ' block-synchronization' : '');
return [
'signature' => [
sprintf('keyId="%s",algorithm="rsa-sha256",headers="%s",signature="dGVzdA=="', $sender->key_id, $list),
],
'block-synchronization' => [
sprintf('url="%s", digest="%s"', $url, $digest),
],
];
}
function bsyncFakeCollection(string $url, array $pairs, string|false|null $digest = null): void
{
[$private] = bsyncKeyPair();
Cache::forever(InstanceActor::PKI_PRIVATE, $private);
$document = [
'@context' => 'https://www.w3.org/ns/activitystreams',
'id' => $url,
'type' => 'OrderedCollection',
'totalItems' => count($pairs),
'orderedItems' => array_map(fn (array $pair) => ['type' => 'Block'] + $pair, $pairs),
];
if ($digest !== false) {
$document['blockSynchronizationDigest'] = $digest ?? BlockSyncService::digest($pairs);
}
Http::fake([
$url => Http::response(json_encode($document), 200, ['Content-Type' => 'application/activity+json']),
]);
}
describe('sender', function () {
it('signs a Block-Synchronization header scoped to each destination', function () {
Http::fake();
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
$carol = bsyncRemoteProfile('remote2.example', 'carol');
bsyncBlock($local, $alice);
bsyncBlock($local, $carol);
bsyncSeedHosts(['joinloops.org', 'remote2.example', 'remote3.example']);
bsyncInProduction(fn () => ActivityPubDeliveryService::pool(
$local,
[
'https://joinloops.org/inbox',
'https://remote2.example/inbox',
'https://remote3.example/inbox',
],
['id' => $local->permalink('#create'), 'type' => 'Create', 'actor' => $local->permalink()]
));
$expected = [
'https://joinloops.org/inbox' => BlockSyncService::digest([bsyncPair($local, $alice)]),
'https://remote2.example/inbox' => BlockSyncService::digest([bsyncPair($local, $carol)]),
'https://remote3.example/inbox' => BlockSyncService::digest([]),
];
Http::assertSentCount(3);
foreach (Http::recorded() as [$request]) {
$params = BlockSyncService::parseHeader($request->header('Block-Synchronization')[0] ?? null);
preg_match('/headers="([^"]*)"/', $request->header('Signature')[0] ?? '', $signed);
expect($params)->not->toBeNull();
expect($params['url'])->toBe(BlockSyncService::endpointUrl());
expect($params['digest'])->toBe($expected[$request->url()]);
expect(explode(' ', $signed[1] ?? ''))->toContain('block-synchronization');
}
});
it('does not reveal anything unless disclosure is enabled', function () {
config(['federation.activitypub.block_sync.disclose' => false]);
Http::fake();
$local = bsyncLocalProfile();
bsyncBlock($local, bsyncRemoteProfile('joinloops.org', 'alice'));
bsyncSeedHosts(['joinloops.org']);
bsyncInProduction(fn () => ActivityPubDeliveryService::pool(
$local,
['https://joinloops.org/inbox'],
['id' => $local->permalink('#create'), 'type' => 'Create', 'actor' => $local->permalink()]
));
Http::assertSent(fn ($request) => empty($request->header('Block-Synchronization')));
$this->get('/f/block_sync')->assertNotFound();
});
it('keeps the cached digests in step with blocks', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncSeedHosts(['joinloops.org']);
expect(BlockSyncService::outboundDigests())->toBe([]);
$filter = UserFilter::create([
'user_id' => $local->id,
'filterable_id' => $alice->id,
'filterable_type' => Profile::class,
'filter_type' => 'block',
]);
expect(BlockSyncService::outboundDigests())->toBe([
'https://joinloops.org' => BlockSyncService::digest([bsyncPair($local, $alice)]),
]);
$filter->delete();
expect(BlockSyncService::outboundDigests())->toBe([]);
});
it('collapses block changes for the same pair into one queued delivery', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncSeedHosts(['joinloops.org']);
$filter = UserFilter::create([
'user_id' => $local->id,
'filterable_id' => $alice->id,
'filterable_type' => Profile::class,
'filter_type' => 'block',
]);
Queue::assertPushed(DeliverBlockActivity::class, 1);
$filter->delete();
Queue::assertPushed(DeliverBlockActivity::class, 1);
});
it('never counts blocks received from other servers as disclosed', function () {
$local = bsyncLocalProfile();
bsyncBlock(bsyncRemoteProfile('joinloops.org', 'alice'), $local);
expect(BlockSyncService::outboundDigests())->toBe([]);
});
});
describe('endpoint', function () {
it('rejects unsigned requests', function () {
$this->get('/f/block_sync', ['Accept' => 'application/activity+json'])->assertStatus(401);
});
it('only lists blocks of actors hosted by the instance that signed the request', function () {
$local = bsyncLocalProfile();
[$private, $public] = bsyncKeyPair();
bsyncRemoteProfile('joinloops.org', 'actor', [
'remote_url' => 'https://joinloops.org/actor',
'key_id' => 'https://joinloops.org/actor#main-key',
'public_key' => $public,
]);
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
$carol = bsyncRemoteProfile('remote2.example', 'carol');
bsyncBlock($local, $alice);
bsyncBlock($local, $carol);
bsyncSeedHosts(['joinloops.org']);
$response = $this->get('/f/block_sync', bsyncSignedGetHeaders($private, 'https://joinloops.org/actor#main-key', '/f/block_sync'))
->assertOk()
->assertJsonPath('type', 'OrderedCollection')
->assertJsonPath('totalItems', 1);
$digest = BlockSyncService::digest([bsyncPair($local, $alice)]);
expect($response->json('orderedItems'))->toBe([['type' => 'Block'] + bsyncPair($local, $alice)]);
expect($response->json('blockSynchronizationDigest'))->toBe($digest);
expect($response->headers->get('ETag'))->toBe('"'.$digest.'"');
});
it('refuses limited peers instead of answering with an empty set', function () {
[$private, $public] = bsyncKeyPair();
bsyncRemoteProfile('joinloops.org', 'actor', [
'remote_url' => 'https://joinloops.org/actor',
'key_id' => 'https://joinloops.org/actor#main-key',
'public_key' => $public,
]);
bsyncSeedHosts(['joinloops.org']);
Cache::put('instances:unlisted:domains', ['joinloops.org'], 1209600);
$this->get('/f/block_sync', bsyncSignedGetHeaders($private, 'https://joinloops.org/actor#main-key', '/f/block_sync'))
->assertForbidden();
});
});
describe('inbox', function () {
it('applies a Block activity and drops follows in both directions', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncSeedHosts(['joinloops.org']);
DB::table('followers')->insert([
['profile_id' => $local->id, 'following_id' => $alice->id, 'created_at' => now(), 'updated_at' => now()],
['profile_id' => $alice->id, 'following_id' => $local->id, 'created_at' => now(), 'updated_at' => now()],
]);
$headers = bsyncInboundHeaders($alice, 'https://joinloops.org/f/block_sync', BlockSyncService::digest([]));
(new Inbox($headers, $local, [
'id' => 'https://joinloops.org/blocks/1',
'type' => 'Block',
'actor' => $alice->remote_url,
'object' => $local->permalink(),
]))->handle();
expect(bsyncBlocks($alice, $local))->toBeTrue();
expect(Follower::whereProfileId($local->id)->whereFollowingId($alice->id)->exists())->toBeFalse();
expect(Follower::whereProfileId($alice->id)->whereFollowingId($local->id)->exists())->toBeFalse();
});
it('removes the block on Undo Block', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
bsyncSeedHosts(['joinloops.org']);
$headers = bsyncInboundHeaders($alice, 'https://joinloops.org/f/block_sync', BlockSyncService::digest([]));
(new Inbox($headers, $local, [
'id' => 'https://joinloops.org/blocks/1/undo',
'type' => 'Undo',
'actor' => $alice->remote_url,
'object' => [
'id' => 'https://joinloops.org/blocks/1',
'type' => 'Block',
'actor' => $alice->remote_url,
'object' => $local->permalink(),
],
]))->handle();
expect(bsyncBlocks($alice, $local))->toBeFalse();
});
it('ignores a Block signed from another server', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
$mallory = bsyncRemoteProfile('evil.example', 'mallory');
bsyncSeedHosts(['joinloops.org', 'evil.example']);
$headers = bsyncInboundHeaders($mallory, 'https://evil.example/f/block_sync', BlockSyncService::digest([]));
(new Inbox($headers, $local, [
'id' => 'https://joinloops.org/blocks/1',
'type' => 'Block',
'actor' => $alice->remote_url,
'object' => $local->permalink(),
]))->handle();
expect(bsyncBlocks($alice, $local))->toBeFalse();
});
});
describe('inbound header', function () {
it('queues a synchronization when the digests differ and remembers the endpoint', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
Instance::updateOrCreate(['domain' => 'joinloops.org']);
bsyncSeedHosts(['joinloops.org']);
BlockSyncService::handleInboundHeaders(bsyncInboundHeaders(
$alice,
'https://joinloops.org/f/block_sync',
BlockSyncService::digest([bsyncPair($alice, $local)])
));
Queue::assertPushed(BlockSyncPipeline::class, 1);
expect(Instance::whereDomain('joinloops.org')->value('block_sync_url'))->toBe('https://joinloops.org/f/block_sync');
});
it('stays quiet when the digests agree', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
bsyncSeedHosts(['joinloops.org']);
BlockSyncService::handleInboundHeaders(bsyncInboundHeaders(
$alice,
'https://joinloops.org/f/block_sync',
BlockSyncService::digest([bsyncPair($alice, $local)])
));
Queue::assertNotPushed(BlockSyncPipeline::class);
});
it('ignores unsigned headers and endpoints on another server', function () {
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncSeedHosts(['joinloops.org']);
BlockSyncService::handleInboundHeaders(bsyncInboundHeaders($alice, 'https://joinloops.org/f/block_sync', str_repeat('1', 64), false));
BlockSyncService::handleInboundHeaders(bsyncInboundHeaders($alice, 'https://victim.example/f/block_sync', str_repeat('1', 64)));
Queue::assertNotPushed(BlockSyncPipeline::class);
});
});
describe('synchronization', function () {
it('adds blocks the authoritative server lists', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, [bsyncPair($alice, $local)]);
$result = BlockSyncService::synchronize('https://joinloops.org', $url, BlockSyncService::digest([bsyncPair($alice, $local)]));
expect($result['added'])->toBe(1);
expect(bsyncBlocks($alice, $local))->toBeTrue();
});
it('removes stale blocks when the collection hashes to its own digest', function () {
$kept = bsyncLocalProfile();
$stale = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $kept);
bsyncBlock($alice, $stale);
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, [bsyncPair($alice, $kept)]);
$result = BlockSyncService::synchronize('https://joinloops.org', $url, BlockSyncService::digest([bsyncPair($alice, $kept)]));
expect($result['status'])->toBe('synchronized');
expect($result['removed'])->toBe(1);
expect(bsyncBlocks($alice, $kept))->toBeTrue();
expect(bsyncBlocks($alice, $stale))->toBeFalse();
});
it('removes nothing when the collection changed while it was fetched', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, [], BlockSyncService::digest([bsyncPair($alice, $local)]));
$result = BlockSyncService::synchronize('https://joinloops.org', $url, str_repeat('1', 64));
expect($result['status'])->toBe('synchronized_without_removals');
expect(bsyncBlocks($alice, $local))->toBeTrue();
});
it('checks removals against the collection, not the header digest', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, []);
$result = BlockSyncService::synchronize('https://joinloops.org', $url, str_repeat('1', 64));
expect($result['status'])->toBe('synchronized');
expect(bsyncBlocks($alice, $local))->toBeFalse();
});
it('still adds blocks from a collection it cannot verify', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, [bsyncPair($alice, $local)], str_repeat('1', 64));
$result = BlockSyncService::synchronize('https://joinloops.org', $url);
expect($result['status'])->toBe('synchronized_without_removals');
expect($result['added'])->toBe(1);
expect(bsyncBlocks($alice, $local))->toBeTrue();
});
it('keeps a block that is younger than the grace period', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local, 1);
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, []);
$result = BlockSyncService::synchronize('https://joinloops.org', $url, BlockSyncService::digest([]));
expect($result['removed'])->toBe(0);
expect(bsyncBlocks($alice, $local))->toBeTrue();
});
it('never treats a failed fetch as an empty collection', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
[$private] = bsyncKeyPair();
Cache::forever(InstanceActor::PKI_PRIVATE, $private);
bsyncSeedHosts(['joinloops.org']);
Http::fake(['*' => Http::response('', 500)]);
$result = BlockSyncService::synchronize('https://joinloops.org', 'https://joinloops.org/f/block_sync', BlockSyncService::digest([]));
expect($result['status'])->toBe('fetch_failed');
expect(bsyncBlocks($alice, $local))->toBeTrue();
});
it('only removes during periodic reconciliation when the collection carries a digest', function () {
$local = bsyncLocalProfile();
$alice = bsyncRemoteProfile('joinloops.org', 'alice');
bsyncBlock($alice, $local);
$url = 'https://joinloops.org/f/block_sync';
bsyncSeedHosts(['joinloops.org']);
bsyncFakeCollection($url, [], false);
expect(BlockSyncService::synchronize('https://joinloops.org', $url)['removed'])->toBe(0);
expect(bsyncBlocks($alice, $local))->toBeTrue();
$url = 'https://joinloops.org/f/block_sync_complete';
bsyncFakeCollection($url, []);
expect(BlockSyncService::synchronize('https://joinloops.org', $url)['removed'])->toBe(1);
expect(bsyncBlocks($alice, $local))->toBeFalse();
});
});

@ -0,0 +1,64 @@
<?php
use App\Services\BlockSyncService;
it('matches the digest example of the FEP', function () {
expect(BlockSyncService::digest([
['actor' => 'https://a.example/users/alice', 'object' => 'https://b.example/users/bob'],
['actor' => 'https://a.example/users/carol', 'object' => 'https://b.example/users/bob'],
]))->toBe('5267015e810867e01494e494c2cbe378fe9bf05e4e5dba51d65f38cbe4a47dd5');
});
it('hashes each pair as actor, space, object', function () {
expect(bin2hex(BlockSyncService::pairHash('https://a.example/users/alice', 'https://b.example/users/bob')))
->toBe('5a0f09e214e8951a292c01ce220e3b768e738ab28917c73a3ced82e8fd8c77c2');
});
it('hashes an empty set to zeros', function () {
expect(BlockSyncService::digest([]))->toBe(str_repeat('0', 64));
});
it('does not depend on the order of the pairs', function () {
$a = ['actor' => 'https://a.example/users/1', 'object' => 'https://b.example/users/1'];
$b = ['actor' => 'https://a.example/users/2', 'object' => 'https://b.example/users/1'];
expect(BlockSyncService::digest([$a, $b]))->toBe(BlockSyncService::digest([$b, $a]));
});
it('parses the example header of the FEP', function () {
expect(BlockSyncService::parseHeader('url="https://a.example/federation/block-sync", digest="5267015e810867e01494e494c2cbe378fe9bf05e4e5dba51d65f38cbe4a47dd5"'))
->toBe([
'url' => 'https://a.example/federation/block-sync',
'digest' => '5267015e810867e01494e494c2cbe378fe9bf05e4e5dba51d65f38cbe4a47dd5',
]);
});
it('rejects malformed headers', function () {
$digest = str_repeat('0', 64);
expect(BlockSyncService::parseHeader(''))->toBeNull();
expect(BlockSyncService::parseHeader('url="a", digest="abc"'))->toBeNull();
expect(BlockSyncService::parseHeader('digest="'.$digest.'"'))->toBeNull();
expect(BlockSyncService::parseHeader('url="a", url="b", digest="'.$digest.'"'))->toBeNull();
});
it('only reads Block items with an actor and an object', function () {
expect(BlockSyncService::itemToPair([
'type' => 'Block',
'actor' => ['id' => 'https://a.example/users/alice'],
'object' => 'https://b.example/users/bob',
]))->toBe(['actor' => 'https://a.example/users/alice', 'object' => 'https://b.example/users/bob']);
expect(BlockSyncService::itemToPair(['type' => 'Follow', 'actor' => 'https://a.example/x', 'object' => 'https://b.example/y']))->toBeNull();
expect(BlockSyncService::itemToPair(['type' => 'Block', 'actor' => 'https://a.example/x']))->toBeNull();
expect(BlockSyncService::itemToPair('https://b.example/users/bob'))->toBeNull();
});
it('reads the collection digest in compact or expanded form', function () {
$digest = '5267015e810867e01494e494c2cbe378fe9bf05e4e5dba51d65f38cbe4a47dd5';
expect(BlockSyncService::collectionDigest(['blockSynchronizationDigest' => strtoupper($digest)]))->toBe($digest);
expect(BlockSyncService::collectionDigest([BlockSyncService::DIGEST_CONTEXT_TERM => $digest]))->toBe($digest);
expect(BlockSyncService::collectionDigest(['blockSynchronizationDigest' => 'abc']))->toBeNull();
expect(BlockSyncService::collectionDigest([]))->toBeNull();
});
Loading…
Cancel
Save