Add basic support for refreshing CSV block/allow lists

app/models/subscribed_advisory.rb

@@ -26,6 +26,9 @@ class SubscribedAdvisory < ApplicationRecord
     domain: 0,
   }, suffix: :target_type
 
+  ACTIONS = actions.keys.freeze
+  TARGET_TYPES = target_types.keys.freeze
+
   before_validation :normalize_target_key
 
   # TODO: handle subdomains

app/services/moderation_subscription_sync_service.rb

@@ -0,0 +1,87 @@
+# frozen_string_literal: true
+
+class ModerationSubscriptionSyncService < BaseService
+  def call(subscription)
+    case subscription.type
+    when 'csv_list'
+      synchronize_csv_moderation_subscription!(subscription)
+    end
+  end
+
+  private
+
+  def synchronize_csv_moderation_subscription!(subscription)
+    Request.new(:get, subscription.url).add_headers('Accept' => 'text/csv').perform do |res|
+      return false unless res.code == 200
+
+      body = res.body_with_limit
+
+      # TODO: find a way to deduplicate with `Admin::Import`
+      csv_converter = lambda do |field, field_info|
+        case field_info.header
+        when '#domain', '#severity'
+          field&.downcase&.strip
+        when '#public_comment'
+          field&.strip
+        when '#reject_media', '#reject_reports', '#obfuscate'
+          ActiveModel::Type::Boolean.new.cast(field&.downcase)
+        else
+          field
+        end
+      end
+
+      csv_data = CSV.new(body, encoding: 'UTF-8', skip_blanks: true, headers: true, converters: csv_converter)
+      csv_data.take(1) # Ensure the headers are read
+      csv_data = CSV.new(body, encoding: 'UTF-8', skip_blanks: true, headers: ['#domain'], converters: csv_converter) unless csv_data.headers&.first == '#domain'
+
+      csv_data.rewind
+      rows = csv_data.take(Admin::Import::ROWS_PROCESSING_LIMIT + 1)
+
+      advisories = rows.filter_map do |row|
+        action = action_from_severity(subscription, row['#severity'])
+        next if action.blank?
+
+        {
+          target_key: TagManager.instance.normalize_domain(row['#domain']),
+          target_type: 'domain',
+          moderation_subscription_id: subscription.id,
+          action: action,
+        }
+      end
+
+      synchronize_advisories(subscription, advisories)
+    end
+  rescue *Mastodon::HTTP_CONNECTION_ERRORS => e
+    Rails.logger.warn "Failed syncing moderation subscription #{subscription.url}: #{e}"
+
+    false
+  end
+
+  def synchronize_advisories(subscription, advisories)
+    SubscribedAdvisory.upsert_all(
+      advisories,
+      unique_by: [:target_type, :target_key, :moderation_subscription_id]
+    )
+
+    SubscribedAdvisory::TARGET_TYPES.each do |target_type|
+      subscription.advisories.where(target_type: target_type).where.not(target_key: advisories.filter_map { |advisory| advisory[:target_key] if advisory[:target_type] == target_type }).delete_all
+    end
+
+    subscription.touch(:last_synced_at)
+
+    true
+  end
+
+  def action_from_severity(subscription, severity)
+    case severity
+    when '', nil
+      subscription.list_action || 'reject'
+    when 'accept', 'allow'
+      'accept' unless subscription.list_action == 'reject'
+    when 'limit', 'silence'
+      'limit' unless subscription.list_action == 'accept'
+    when 'block', 'reject', 'suspend'
+      'reject' unless subscription.list_action == 'accept'
+    end
+  end
+end

spec/services/moderation_subscription_sync_service_spec.rb

@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ModerationSubscriptionSyncService do
+  subject { described_class.new }
+
+  context 'with a CSV list' do
+    let(:moderation_subscription) { Fabricate(:moderation_subscription, type: :csv_list) }
+
+    context 'when the CSV list returns an error' do
+      before do
+        stub_request(:get, moderation_subscription.url)
+          .to_return(status: 500)
+      end
+
+      it 'does not update the list' do
+        expect { subject.call(moderation_subscription) }
+          .to not_change(moderation_subscription, :updated_at)
+          .and not_change(moderation_subscription, :advisories)
+      end
+    end
+
+    context 'when the CSV list is only domains' do
+      let(:raw_csv) do
+        <<~CSV
+          evil.com
+          example.com
+        CSV
+      end
+
+      before do
+        moderation_subscription.advisories.create!(target_type: :domain, target_key: 'benign.com', action: :reject)
+
+        stub_request(:get, moderation_subscription.url)
+          .to_return(status: 200, headers: { 'Content-Type': 'text/csv' }, body: raw_csv)
+      end
+
+      it 'updates the list accordingly' do
+        expect { subject.call(moderation_subscription) }
+          .to change(moderation_subscription, :last_synced_at)
+
+        expect(moderation_subscription.advisories.pluck(:target_type, :target_key, :action))
+          .to contain_exactly(['domain', 'example.com', 'reject'], ['domain', 'evil.com', 'reject'])
+      end
+    end
+
+    context 'when the CSV list has a header' do
+      let(:raw_csv) do
+        <<~CSV
+          #domain,#severity
+          evil.com,suspend
+          suspended.example.com,suspend
+          silenced.example.com,silence
+          allowed.example.com,allow
+        CSV
+      end
+
+      before do
+        moderation_subscription.advisories.create!(target_type: :domain, target_key: 'benign.com', action: :reject)
+
+        stub_request(:get, moderation_subscription.url)
+          .to_return(status: 200, headers: { 'Content-Type': 'text/csv' }, body: raw_csv)
+      end
+
+      it 'updates the list accordingly' do
+        expect { subject.call(moderation_subscription) }
+          .to change(moderation_subscription, :last_synced_at)
+
+        expect(moderation_subscription.advisories.pluck(:target_type, :target_key, :action))
+          .to contain_exactly(
+            ['domain', 'suspended.example.com', 'reject'],
+            ['domain', 'silenced.example.com', 'limit'],
+            ['domain', 'evil.com', 'reject']
+          )
+      end
+    end
+  end
+end