mirror of
				https://github.com/superseriousbusiness/gotosocial.git
				synced 2025-11-03 21:32:24 -06:00 
			
		
		
		
	* start replacing client + federator + media workers with new worker + queue types
* refactor federatingDB.Delete(), drop queued messages when deleting account / status
* move all queue purging to the processor workers
* undo toolchain updates
* code comments, ensure dereferencer worker pool gets started
* update gruf libraries in readme
* start the job scheduler separately to the worker pools
* reshuffle ordering or server.go + remove duplicate worker start / stop
* update go-list version
* fix vendoring
* move queue invalidation to before wipeing / deletion, to ensure queued work not dropped
* add logging to worker processing functions in testrig, don't start workers in unexpected places
* update go-structr to add (+then rely on) QueueCtx{} type
* ensure more worker pools get started properly in tests
* fix remaining broken tests relying on worker queue logic
* fix account test suite queue popping logic, ensure noop workers do not pull from queue
* move back accidentally shuffled account deletion order
* ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete()
* silently drop deletes from accounts not permitted to
* don't warn log on forwarded deletes
* make if else clauses easier to parse
* use getFederatorMsg()
* improved code comment
* improved code comment re: requesting account delete checks
* remove boolean result from worker start / stop since false = already running or already stopped
* remove optional passed-in http.client
* remove worker starting from the admin CLI commands (we don't need to handle side-effects)
* update prune cli to start scheduler but not all of the workers
* fix rebase issues
* remove redundant return statements
* i'm sorry sir linter
		
	
			
		
			
				
	
	
		
			350 lines
		
	
	
	
		
			11 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			350 lines
		
	
	
	
		
			11 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
// GoToSocial
 | 
						|
// Copyright (C) GoToSocial Authors admin@gotosocial.org
 | 
						|
// SPDX-License-Identifier: AGPL-3.0-or-later
 | 
						|
//
 | 
						|
// This program is free software: you can redistribute it and/or modify
 | 
						|
// it under the terms of the GNU Affero General Public License as published by
 | 
						|
// the Free Software Foundation, either version 3 of the License, or
 | 
						|
// (at your option) any later version.
 | 
						|
//
 | 
						|
// This program is distributed in the hope that it will be useful,
 | 
						|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
 | 
						|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 | 
						|
// GNU Affero General Public License for more details.
 | 
						|
//
 | 
						|
// You should have received a copy of the GNU Affero General Public License
 | 
						|
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
 | 
						|
 | 
						|
package admin
 | 
						|
 | 
						|
import (
 | 
						|
	"context"
 | 
						|
	"errors"
 | 
						|
	"fmt"
 | 
						|
	"time"
 | 
						|
 | 
						|
	"codeberg.org/gruf/go-kv"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/ap"
 | 
						|
	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/config"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/db"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/id"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/log"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/messages"
 | 
						|
	"github.com/superseriousbusiness/gotosocial/internal/text"
 | 
						|
)
 | 
						|
 | 
						|
func (p *Processor) createDomainBlock(
 | 
						|
	ctx context.Context,
 | 
						|
	adminAcct *gtsmodel.Account,
 | 
						|
	domain string,
 | 
						|
	obfuscate bool,
 | 
						|
	publicComment string,
 | 
						|
	privateComment string,
 | 
						|
	subscriptionID string,
 | 
						|
) (*apimodel.DomainPermission, string, gtserror.WithCode) {
 | 
						|
	// Check if a block already exists for this domain.
 | 
						|
	domainBlock, err := p.state.DB.GetDomainBlock(ctx, domain)
 | 
						|
	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 | 
						|
		// Something went wrong in the DB.
 | 
						|
		err = gtserror.Newf("db error getting domain block %s: %w", domain, err)
 | 
						|
		return nil, "", gtserror.NewErrorInternalError(err)
 | 
						|
	}
 | 
						|
 | 
						|
	if domainBlock == nil {
 | 
						|
		// No block exists yet, create it.
 | 
						|
		domainBlock = >smodel.DomainBlock{
 | 
						|
			ID:                 id.NewULID(),
 | 
						|
			Domain:             domain,
 | 
						|
			CreatedByAccountID: adminAcct.ID,
 | 
						|
			PrivateComment:     text.SanitizeToPlaintext(privateComment),
 | 
						|
			PublicComment:      text.SanitizeToPlaintext(publicComment),
 | 
						|
			Obfuscate:          &obfuscate,
 | 
						|
			SubscriptionID:     subscriptionID,
 | 
						|
		}
 | 
						|
 | 
						|
		// Insert the new block into the database.
 | 
						|
		if err := p.state.DB.CreateDomainBlock(ctx, domainBlock); err != nil {
 | 
						|
			err = gtserror.Newf("db error putting domain block %s: %w", domain, err)
 | 
						|
			return nil, "", gtserror.NewErrorInternalError(err)
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	actionID := id.NewULID()
 | 
						|
 | 
						|
	// Process domain block side
 | 
						|
	// effects asynchronously.
 | 
						|
	if errWithCode := p.actions.Run(
 | 
						|
		ctx,
 | 
						|
		>smodel.AdminAction{
 | 
						|
			ID:             actionID,
 | 
						|
			TargetCategory: gtsmodel.AdminActionCategoryDomain,
 | 
						|
			TargetID:       domain,
 | 
						|
			Type:           gtsmodel.AdminActionSuspend,
 | 
						|
			AccountID:      adminAcct.ID,
 | 
						|
			Text:           domainBlock.PrivateComment,
 | 
						|
		},
 | 
						|
		func(ctx context.Context) gtserror.MultiError {
 | 
						|
			// Log start + finish.
 | 
						|
			l := log.WithFields(kv.Fields{
 | 
						|
				{"domain", domain},
 | 
						|
				{"actionID", actionID},
 | 
						|
			}...).WithContext(ctx)
 | 
						|
 | 
						|
			skip, err := p.skipBlockSideEffects(ctx, domain)
 | 
						|
			if err != nil {
 | 
						|
				return err
 | 
						|
			}
 | 
						|
			if skip != "" {
 | 
						|
				l.Infof("skipping domain block side effects: %s", skip)
 | 
						|
				return nil
 | 
						|
			}
 | 
						|
 | 
						|
			l.Info("processing domain block side effects")
 | 
						|
			defer func() { l.Info("finished processing domain block side effects") }()
 | 
						|
 | 
						|
			return p.domainBlockSideEffects(ctx, domainBlock)
 | 
						|
		},
 | 
						|
	); errWithCode != nil {
 | 
						|
		return nil, actionID, errWithCode
 | 
						|
	}
 | 
						|
 | 
						|
	apiDomainBlock, errWithCode := p.apiDomainPerm(ctx, domainBlock, false)
 | 
						|
	if errWithCode != nil {
 | 
						|
		return nil, actionID, errWithCode
 | 
						|
	}
 | 
						|
 | 
						|
	return apiDomainBlock, actionID, nil
 | 
						|
}
 | 
						|
 | 
						|
// skipBlockSideEffects checks if side effects of block creation
 | 
						|
// should be skipped for the given domain, taking account of
 | 
						|
// instance federation mode, and existence of any allows
 | 
						|
// which ought to "shield" this domain from being blocked.
 | 
						|
//
 | 
						|
// If the caller should skip, the returned string will be non-zero
 | 
						|
// and will be set to a reason why side effects should be skipped.
 | 
						|
//
 | 
						|
//   - blocklist mode + allow exists: "..." (skip)
 | 
						|
//   - blocklist mode + no allow:     ""    (don't skip)
 | 
						|
//   - allowlist mode + allow exists: ""    (don't skip)
 | 
						|
//   - allowlist mode + no allow:     ""    (don't skip)
 | 
						|
func (p *Processor) skipBlockSideEffects(
 | 
						|
	ctx context.Context,
 | 
						|
	domain string,
 | 
						|
) (string, gtserror.MultiError) {
 | 
						|
	var (
 | 
						|
		skip string // Assume "" (don't skip).
 | 
						|
		errs gtserror.MultiError
 | 
						|
	)
 | 
						|
 | 
						|
	// Never skip block side effects in allowlist mode.
 | 
						|
	fediMode := config.GetInstanceFederationMode()
 | 
						|
	if fediMode == config.InstanceFederationModeAllowlist {
 | 
						|
		return skip, errs
 | 
						|
	}
 | 
						|
 | 
						|
	// We know we're in blocklist mode.
 | 
						|
	//
 | 
						|
	// We want to skip domain block side
 | 
						|
	// effects if an allow is already
 | 
						|
	// in place which overrides the block.
 | 
						|
 | 
						|
	// Check if an explicit allow exists for this domain.
 | 
						|
	domainAllow, err := p.state.DB.GetDomainAllow(ctx, domain)
 | 
						|
	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 | 
						|
		errs.Appendf("error getting domain allow: %w", err)
 | 
						|
		return skip, errs
 | 
						|
	}
 | 
						|
 | 
						|
	if domainAllow != nil {
 | 
						|
		skip = "running in blocklist mode, and an explicit allow exists for this domain"
 | 
						|
		return skip, errs
 | 
						|
	}
 | 
						|
 | 
						|
	return skip, errs
 | 
						|
}
 | 
						|
 | 
						|
// domainBlockSideEffects processes the side effects of a domain block:
 | 
						|
//
 | 
						|
//  1. Strip most info away from the instance entry for the domain.
 | 
						|
//  2. Pass each account from the domain to the processor for deletion.
 | 
						|
//
 | 
						|
// It should be called asynchronously, since it can take a while when
 | 
						|
// there are many accounts present on the given domain.
 | 
						|
func (p *Processor) domainBlockSideEffects(
 | 
						|
	ctx context.Context,
 | 
						|
	block *gtsmodel.DomainBlock,
 | 
						|
) gtserror.MultiError {
 | 
						|
	var errs gtserror.MultiError
 | 
						|
 | 
						|
	// If we have an instance entry for this domain,
 | 
						|
	// update it with the new block ID and clear all fields
 | 
						|
	instance, err := p.state.DB.GetInstance(ctx, block.Domain)
 | 
						|
	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 | 
						|
		errs.Appendf("db error getting instance %s: %w", block.Domain, err)
 | 
						|
		return errs
 | 
						|
	}
 | 
						|
 | 
						|
	if instance != nil {
 | 
						|
		// We had an entry for this domain.
 | 
						|
		columns := stubbifyInstance(instance, block.ID)
 | 
						|
		if err := p.state.DB.UpdateInstance(ctx, instance, columns...); err != nil {
 | 
						|
			errs.Appendf("db error updating instance: %w", err)
 | 
						|
			return errs
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	// For each account that belongs to this domain,
 | 
						|
	// process an account delete message to remove
 | 
						|
	// that account's posts, media, etc.
 | 
						|
	if err := p.rangeDomainAccounts(ctx, block.Domain, func(account *gtsmodel.Account) {
 | 
						|
		if err := p.state.Workers.Client.Process(ctx, &messages.FromClientAPI{
 | 
						|
			APObjectType:   ap.ActorPerson,
 | 
						|
			APActivityType: ap.ActivityDelete,
 | 
						|
			GTSModel:       block,
 | 
						|
			Origin:         account,
 | 
						|
			Target:         account,
 | 
						|
		}); err != nil {
 | 
						|
			errs.Append(err)
 | 
						|
		}
 | 
						|
	}); err != nil {
 | 
						|
		errs.Appendf("db error ranging through accounts: %w", err)
 | 
						|
	}
 | 
						|
 | 
						|
	return errs
 | 
						|
}
 | 
						|
 | 
						|
func (p *Processor) deleteDomainBlock(
 | 
						|
	ctx context.Context,
 | 
						|
	adminAcct *gtsmodel.Account,
 | 
						|
	domainBlockID string,
 | 
						|
) (*apimodel.DomainPermission, string, gtserror.WithCode) {
 | 
						|
	domainBlock, err := p.state.DB.GetDomainBlockByID(ctx, domainBlockID)
 | 
						|
	if err != nil {
 | 
						|
		if !errors.Is(err, db.ErrNoEntries) {
 | 
						|
			// Real error.
 | 
						|
			err = gtserror.Newf("db error getting domain block: %w", err)
 | 
						|
			return nil, "", gtserror.NewErrorInternalError(err)
 | 
						|
		}
 | 
						|
 | 
						|
		// There are just no entries for this ID.
 | 
						|
		err = fmt.Errorf("no domain block entry exists with ID %s", domainBlockID)
 | 
						|
		return nil, "", gtserror.NewErrorNotFound(err, err.Error())
 | 
						|
	}
 | 
						|
 | 
						|
	// Prepare the domain block to return, *before* the deletion goes through.
 | 
						|
	apiDomainBlock, errWithCode := p.apiDomainPerm(ctx, domainBlock, false)
 | 
						|
	if errWithCode != nil {
 | 
						|
		return nil, "", errWithCode
 | 
						|
	}
 | 
						|
 | 
						|
	// Delete the original domain block.
 | 
						|
	if err := p.state.DB.DeleteDomainBlock(ctx, domainBlock.Domain); err != nil {
 | 
						|
		err = gtserror.Newf("db error deleting domain block: %w", err)
 | 
						|
		return nil, "", gtserror.NewErrorInternalError(err)
 | 
						|
	}
 | 
						|
 | 
						|
	actionID := id.NewULID()
 | 
						|
 | 
						|
	// Process domain unblock side
 | 
						|
	// effects asynchronously.
 | 
						|
	if errWithCode := p.actions.Run(
 | 
						|
		ctx,
 | 
						|
		>smodel.AdminAction{
 | 
						|
			ID:             actionID,
 | 
						|
			TargetCategory: gtsmodel.AdminActionCategoryDomain,
 | 
						|
			TargetID:       domainBlock.Domain,
 | 
						|
			Type:           gtsmodel.AdminActionUnsuspend,
 | 
						|
			AccountID:      adminAcct.ID,
 | 
						|
		},
 | 
						|
		func(ctx context.Context) gtserror.MultiError {
 | 
						|
			// Log start + finish.
 | 
						|
			l := log.WithFields(kv.Fields{
 | 
						|
				{"domain", domainBlock.Domain},
 | 
						|
				{"actionID", actionID},
 | 
						|
			}...).WithContext(ctx)
 | 
						|
 | 
						|
			l.Info("processing domain unblock side effects")
 | 
						|
			defer func() { l.Info("finished processing domain unblock side effects") }()
 | 
						|
 | 
						|
			return p.domainUnblockSideEffects(ctx, domainBlock)
 | 
						|
		},
 | 
						|
	); errWithCode != nil {
 | 
						|
		return nil, actionID, errWithCode
 | 
						|
	}
 | 
						|
 | 
						|
	return apiDomainBlock, actionID, nil
 | 
						|
}
 | 
						|
 | 
						|
// domainUnblockSideEffects processes the side effects of undoing a
 | 
						|
// domain block:
 | 
						|
//
 | 
						|
//  1. Mark instance entry as no longer suspended.
 | 
						|
//  2. Mark each account from the domain as no longer suspended, if the
 | 
						|
//     suspension origin corresponds to the ID of the provided domain block.
 | 
						|
//
 | 
						|
// It should be called asynchronously, since it can take a while when
 | 
						|
// there are many accounts present on the given domain.
 | 
						|
func (p *Processor) domainUnblockSideEffects(
 | 
						|
	ctx context.Context,
 | 
						|
	block *gtsmodel.DomainBlock,
 | 
						|
) gtserror.MultiError {
 | 
						|
	var errs gtserror.MultiError
 | 
						|
 | 
						|
	// Update instance entry for this domain, if we have it.
 | 
						|
	instance, err := p.state.DB.GetInstance(ctx, block.Domain)
 | 
						|
	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 | 
						|
		errs.Appendf("db error getting instance %s: %w", block.Domain, err)
 | 
						|
	}
 | 
						|
 | 
						|
	if instance != nil {
 | 
						|
		// We had an entry, update it to signal
 | 
						|
		// that it's no longer suspended.
 | 
						|
		instance.SuspendedAt = time.Time{}
 | 
						|
		instance.DomainBlockID = ""
 | 
						|
		if err := p.state.DB.UpdateInstance(
 | 
						|
			ctx,
 | 
						|
			instance,
 | 
						|
			"suspended_at",
 | 
						|
			"domain_block_id",
 | 
						|
		); err != nil {
 | 
						|
			errs.Appendf("db error updating instance: %w", err)
 | 
						|
			return errs
 | 
						|
		}
 | 
						|
	}
 | 
						|
 | 
						|
	// Unsuspend all accounts whose suspension origin was this domain block.
 | 
						|
	if err := p.rangeDomainAccounts(ctx, block.Domain, func(account *gtsmodel.Account) {
 | 
						|
		if account.SuspensionOrigin == "" || account.SuspendedAt.IsZero() {
 | 
						|
			// Account wasn't suspended, nothing to do.
 | 
						|
			return
 | 
						|
		}
 | 
						|
 | 
						|
		if account.SuspensionOrigin != block.ID {
 | 
						|
			// Account was suspended, but not by
 | 
						|
			// this domain block, leave it alone.
 | 
						|
			return
 | 
						|
		}
 | 
						|
 | 
						|
		// Account was suspended by this domain
 | 
						|
		// block, mark it as unsuspended.
 | 
						|
		account.SuspendedAt = time.Time{}
 | 
						|
		account.SuspensionOrigin = ""
 | 
						|
 | 
						|
		if err := p.state.DB.UpdateAccount(
 | 
						|
			ctx,
 | 
						|
			account,
 | 
						|
			"suspended_at",
 | 
						|
			"suspension_origin",
 | 
						|
		); err != nil {
 | 
						|
			errs.Appendf("db error updating account %s: %w", account.Username, err)
 | 
						|
		}
 | 
						|
	}); err != nil {
 | 
						|
		errs.Appendf("db error ranging through accounts: %w", err)
 | 
						|
	}
 | 
						|
 | 
						|
	return errs
 | 
						|
}
 |