Refactor Geo Transfer and implement generic blob transfer

Extracted JWT logic to the API files intead of multiple levels of
service class.

ee/app/services/geo/blob_upload_service.rb

+# frozen_string_literal: true
+
+module Geo
+  class BlobUploadService
+    include ExclusiveLeaseGuard
+    include ::Gitlab::Geo::LogHelpers
+
+    attr_reader :replicable_name, :blob_id, :decoded_params, :replicator
+
+    def initialize(replicable_name:, blob_id:, decoded_params:)
+      @replicable_name = replicable_name
+      @blob_id = blob_id
+      @decoded_params = decoded_params
+      @replicator = Gitlab::Geo::Replicator.for_replicable_name(params[:replicable_name])
+
+      fail_unimplemented!(replicable_name) unless @replicator
+    end
+
+    def execute
+      retriever.execute
+    end
+
+    def retriever
+      retriever = Gitlab::Geo::Replication::BlobRetriever.new(replicable_name, blob_id)
+    end
+
+    private
+
+    def fail_unimplemented!(replicable_name)
+      error_message = "Cannot find a Geo::Replicator for #{replicable_name}"
+
+      log_error(error_message)
+
+      raise NotImplementedError, error_message
+    end
+
+    # This is called by LogHelpers to build json log with context info
+    #
+    # @see ::Gitlab::Geo::LogHelpers
+    def extra_log_data
+      {
+        replicable_name: replicable_name,
+        blob_id: blob_id
+      }.compact
+    end
+  end
+end

ee/app/services/geo/file_upload_service.rb

@@ -5,37 +5,25 @@ module Geo
   #   * Handling file requests from the secondary over the API
   #   * Returning the necessary response data to send the file back
   class FileUploadService < BaseFileService
-    attr_reader :auth_header
-    include ::Gitlab::Utils::StrongMemoize
+    attr_reader :decoded_params
 
-    def initialize(params, auth_header)
+    def initialize(params, decoded_params)
       super(params[:type], params[:id])
-      @auth_header = auth_header
+
+      @decoded_params = decoded_params
     end
 
     # Returns { code: :ok, file: CarrierWave File object } upon success
     def execute
-      return unless decoded_authorization.present? && jwt_scope_valid?
-
       retriever.execute
     end
 
     def retriever
-      retriever_klass.new(object_db_id, decoded_authorization)
+      retriever_klass.new(object_db_id, decoded_params)
     end
 
     private
 
-    def jwt_scope_valid?
-      (decoded_authorization[:file_type] == object_type.to_s) && (decoded_authorization[:file_id] == object_db_id)
-    end
-
-    def decoded_authorization
-      strong_memoize(:decoded_authorization) do
-        ::Gitlab::Geo::JwtRequestDecoder.new(auth_header).decode
-      end
-    end
-
     def retriever_klass
       return Gitlab::Geo::Replication::FileRetriever if user_upload?
       return Gitlab::Geo::Replication::JobArtifactRetriever if job_artifact?

ee/lib/api/geo.rb

@@ -11,6 +11,32 @@ module API
           valid_attributes = params.keys & allowed_attributes
           params.slice(*valid_attributes)
         end
+
+        def jwt_decoder
+          ::Gitlab::Geo::JwtRequestDecoder.new(headers['Authorization'])
+        end
+
+        # Check if a Geo request is legit or fail the flow
+        #
+        # @param [Hash] attributes to be matched against JWT
+        def authorize_geo_transfer!(**attributes)
+          unauthorized! unless jwt_decoder.valid_attributes?(**attributes)
+        end
+      end
+
+      params do
+        requires :replicable_name, type: String, desc: 'Replicable name (eg. package_file)'
+        requires :id, type: Integer, desc: 'The model ID that needs to be transferred'
+      end
+      get 'retrieve/:replicable_name/:id' do
+        check_gitlab_geo_request_ip!
+        authorize_geo_transfer!(replicable_name: params[:replicable_name], id: params[:id])
+
+        decoded_params = jwt_decoder.decode
+        service = Geo::BlobUploadService.new(replicable_name: params[:replicable_name],
+                                             blob_id: params[:id],
+                                             decoded_params: decoded_params)
+        service.execute
       end
 
       # Verify the GitLab Geo transfer request is valid
@@ -27,12 +53,12 @@ module API
       end
       get 'transfers/:type/:id' do
         check_gitlab_geo_request_ip!
+        authorize_geo_transfer!(file_type: params[:type], file_id: params[:id])
 
-        service = ::Geo::FileUploadService.new(params, headers['Authorization'])
+        decoded_params = jwt_decoder.decode
+        service = ::Geo::FileUploadService.new(params, decoded_params)
         response = service.execute
 
-        unauthorized! unless response.present?
-
         if response[:code] == :ok
           file = response[:file]
           present_carrierwave_file!(file)

ee/lib/gitlab/geo/jwt_request_decoder.rb

@@ -4,6 +4,7 @@ module Gitlab
   module Geo
     class JwtRequestDecoder
       include LogHelpers
+      include ::Gitlab::Utils::StrongMemoize
 
       IAT_LEEWAY = 60.seconds.to_i
 
@@ -18,8 +19,25 @@ module Gitlab
         @auth_header = auth_header
       end
 
+      # Return decoded attributes from informed header
+      #
+      # @return [Hash] decoded attributes
       def decode
-        decode_geo_request
+        strong_memoize(:decoded_authorization) do
+          decode_geo_request
+        end
+      end
+
+      # Check if set of attributes match against attributes decoded from JWT
+      #
+      # @param [Hash] attributes to be matched against JWT
+      # @return bool true
+      def valid_attributes?(**attributes)
+        decoded_attributes = decode
+
+        return false if decoded_attributes.nil?
+
+        attributes.all? { |attribute, value| decoded_attributes[attribute] == value }
       end
 
       private
@@ -72,6 +90,7 @@ module Gitlab
                            geo_node&.secret_access_key
                          end
       end
+
       # rubocop: enable CodeReuse/ActiveRecord
 
       def decode_auth_header

ee/lib/gitlab/geo/replication/file_retriever.rb

@@ -11,7 +11,8 @@ module Gitlab
         def execute
           return error('Upload not found') unless recorded_file
           return file_not_found(recorded_file) unless recorded_file.exist?
-          return error('Upload not found') unless valid?
+          return error('Invalid request') unless valid?
+          return error('Checksum mismatch') unless matches_checksum?
 
           success(recorded_file.retrieve_uploader)
         end
@@ -25,10 +26,8 @@ module Gitlab
         end
 
         def valid?
-          matches_requested_model? && matches_checksum?
-        end
+          return false if message.nil?
 
-        def matches_requested_model?
           message[:id] == recorded_file.model_id &&
             message[:type] == recorded_file.model_type
         end

ee/lib/gitlab/geo/replication/job_artifact_retriever.rb

@@ -9,9 +9,8 @@ module Gitlab
       #
       class JobArtifactRetriever < BaseRetriever
         def execute
-          unless job_artifact.present?
-            return error('Job artifact not found')
-          end
+          return error('Job artifact not found') unless job_artifact.present?
+
 
           unless job_artifact.file.present? && job_artifact.file.exists?
             log_error("Could not upload job artifact because it does not have a file", id: job_artifact.id)

ee/lib/gitlab/geo/replication/lfs_retriever.rb

@@ -9,6 +9,7 @@ module Gitlab
       #
       class LfsRetriever < BaseRetriever
         def execute
+          return error('Invalid request') unless valid?
           return error('LFS object not found') unless lfs_object
           return error('LFS object not found') unless matches_checksum?
 
@@ -29,6 +30,10 @@ module Gitlab
           end
         end
 
+        def valid?
+          !message.nil?
+        end
+
         def matches_checksum?
           message[:checksum] == lfs_object.oid
         end

ee/spec/lib/gitlab/geo/jwt_request_decoder_spec.rb

@@ -6,7 +6,7 @@ describe Gitlab::Geo::JwtRequestDecoder do
   include EE::GeoHelpers
 
   let!(:primary_node) { FactoryBot.create(:geo_node, :primary) }
-  let(:data) { { input: 123 } }
+  let(:data) { { input: 123, other_input: 'string value' } }
   let(:request) { Gitlab::Geo::TransferRequest.new(data) }
 
   subject { described_class.new(request.headers['Authorization']) }
@@ -60,4 +60,18 @@ describe Gitlab::Geo::JwtRequestDecoder do
       expect { subject.decode }.to raise_error(Gitlab::Geo::InvalidDecryptionKeyError)
     end
   end
+
+  describe '#valid_attributes?' do
+    it 'returns true when all informed attributes and decoded data are all the same' do
+      expect(subject.valid_attributes?(input: 123, other_input: 'string value')).to be_truthy
+    end
+
+    it 'returns true when informed attributes is a slice of decoded data' do
+      expect(subject.valid_attributes?(input: 123)).to be_truthy
+    end
+
+    it 'returns false when one informed data doesnt match its corresponding decoded one' do
+      expect(subject.valid_attributes?(input: 123, other_input: 'wrong value')).to be_falsey
+    end
+  end
 end

ee/spec/lib/gitlab/geo/replication/file_retriever_spec.rb

@@ -28,7 +28,7 @@ describe Gitlab::Geo::Replication::FileRetriever, :geo do
           let(:message) { { id: 10000, type: upload.model_type, checksum: 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855' } }
 
           it 'returns an error hash' do
-            expect(subject).to include(code: :not_found, message: "Upload not found")
+            expect(subject).to include(code: :not_found, message: 'Invalid request')
           end
         end
 
@@ -36,7 +36,7 @@ describe Gitlab::Geo::Replication::FileRetriever, :geo do
           let(:message) { { id: upload.model_id, type: 'bad_type', checksum: 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855' } }
 
           it 'returns an error hash' do
-            expect(subject).to include(code: :not_found, message: "Upload not found")
+            expect(subject).to include(code: :not_found, message: 'Invalid request')
           end
         end
 
@@ -44,7 +44,7 @@ describe Gitlab::Geo::Replication::FileRetriever, :geo do
           let(:message) { { id: upload.model_id, type: upload.model_type, checksum: 'doesnotmatch' } }
 
           it 'returns an error hash' do
-            expect(subject).to include(code: :not_found, message: "Upload not found")
+            expect(subject).to include(code: :not_found, message: 'Checksum mismatch')
           end
         end
       end
@@ -62,7 +62,7 @@ describe Gitlab::Geo::Replication::FileRetriever, :geo do
       it 'returns an error hash' do
         result = described_class.new(10000, {}).execute
 
-        expect(result).to eq(code: :not_found, message: "Upload not found")
+        expect(result).to eq(code: :not_found, message: 'Upload not found')
       end
     end
   end

ee/spec/services/geo/blob_upload_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::BlobUploadService do
+  include ::EE::GeoHelpers
+
+  let_it_be(:primary) { create(:geo_node, :primary) }
+  let_it_be(:secondary) { create(:geo_node) }
+
+  subject { described_class.new() }
+end

ee/spec/services/geo/file_upload_service_spec.rb

@@ -7,9 +7,6 @@ describe Geo::FileUploadService do
 
   let_it_be(:node) { create(:geo_node, :primary) }
 
-  let(:transfer_request) { Gitlab::Geo::TransferRequest.new(request_data) }
-  let(:req_header) { transfer_request.headers['Authorization'] }
-
   before do
     stub_current_geo_node(node)
   end
@@ -36,23 +33,13 @@ describe Geo::FileUploadService do
     end
   end
 
-  shared_examples 'no authorization header' do
-    it 'returns nil' do
+  shared_examples 'no decoded params' do
+    it 'returns invalid request error' do
       service = described_class.new(params, nil)
 
-      expect(service.execute).to be_nil
-    end
-  end
-
-  shared_examples 'wrong scope' do
-    context 'at least one scope parameter is wrong' do
-      let(:transfer_request) { Gitlab::Geo::TransferRequest.new(request_data.merge(file_type: 'wrong')) }
-
-      it 'returns nil' do
-        service = described_class.new(params, req_header)
-
-        expect(service.execute).to be_nil
-      end
+      response = service.execute
+      expect(response[:code]).to eq(:not_found)
+      expect(response[:message]).to eq('Invalid request')
     end
   end
 
@@ -64,7 +51,7 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::FileTransfer.new(:avatar, upload).request_data }
 
       it 'sends avatar file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -72,8 +59,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to eq(user.avatar.path)
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'group avatar' do
@@ -83,7 +69,7 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::FileTransfer.new(:avatar, upload).request_data }
 
       it 'sends avatar file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -91,8 +77,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to eq(group.avatar.path)
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'project avatar' do
@@ -102,7 +87,7 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::FileTransfer.new(:avatar, upload).request_data }
 
       it 'sends avatar file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -110,8 +95,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to eq(project.avatar.path)
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'attachment' do
@@ -121,7 +105,7 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::FileTransfer.new(:attachment, upload).request_data }
 
       it 'sends attachment file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -129,8 +113,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to eq(note.attachment.path)
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'file upload' do
@@ -145,7 +128,7 @@ describe Geo::FileUploadService do
       end
 
       it 'sends the file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -153,8 +136,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to end_with('dk.png')
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'namespace file upload' do
@@ -169,7 +151,7 @@ describe Geo::FileUploadService do
       end
 
       it 'sends the file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -177,8 +159,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to end_with('dk.png')
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'LFS Object' do
@@ -187,7 +168,7 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::LfsTransfer.new(lfs_object).request_data }
 
       it 'sends LFS file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -195,8 +176,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to eq(lfs_object.file.path)
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
 
     context 'job artifact' do
@@ -205,16 +185,13 @@ describe Geo::FileUploadService do
       let(:request_data) { Gitlab::Geo::Replication::JobArtifactTransfer.new(job_artifact).request_data }
 
       it 'sends job artifact file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
         expect(response[:code]).to eq(:ok)
         expect(response[:file].path).to eq(job_artifact.file.path)
       end
-
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
     end
 
     context 'import export archive' do
@@ -229,7 +206,7 @@ describe Geo::FileUploadService do
       end
 
       it 'sends the file' do
-        service = described_class.new(params, req_header)
+        service = described_class.new(params, request_data)
 
         response = service.execute
 
@@ -237,8 +214,7 @@ describe Geo::FileUploadService do
         expect(response[:file].path).to end_with('tar.gz')
       end
 
-      include_examples 'no authorization header'
-      include_examples 'wrong scope'
+      include_examples 'no decoded params'
     end
   end
 end