Add verification header for streamed events

Streamed audit events now include a new
HTTP header to allow event consumers to validate
the event origin.

Changelog: added

db/migrate/20220105082217_add_verification_token_to_external_ae_destinations.rb

+# frozen_string_literal: true
+
+class AddVerificationTokenToExternalAeDestinations < Gitlab::Database::Migration[1.0]
+  def up
+    # rubocop:disable Migration/AddLimitToTextColumns
+    add_column :audit_events_external_audit_event_destinations, :verification_token, :text
+    # rubocop:enable Migration/AddLimitToTextColumns
+  end
+
+  def down
+    remove_column :audit_events_external_audit_event_destinations, :verification_token
+  end
+end

db/migrate/20220117082611_add_text_limit_to_exad_verification_tokens.rb

+# frozen_string_literal: true
+
+class AddTextLimitToExadVerificationTokens < Gitlab::Database::Migration[1.0]
+  disable_ddl_transaction!
+
+  def up
+    add_text_limit :audit_events_external_audit_event_destinations, :verification_token, 24
+  end
+
+  def down
+    remove_text_limit :audit_events_external_audit_event_destinations, :verification_token
+  end
+end

db/migrate/20220119094023_add_unique_index_to_aed_verification_token.rb

+# frozen_string_literal: true
+
+class AddUniqueIndexToAedVerificationToken < Gitlab::Database::Migration[1.0]
+  disable_ddl_transaction!
+
+  INDEX_NAME = 'index_audit_events_external_audit_on_verification_token'
+
+  def up
+    add_concurrent_index :audit_events_external_audit_event_destinations, :verification_token, unique: true, name: INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index_by_name :audit_events_external_audit_event_destinations, INDEX_NAME
+  end
+end

db/post_migrate/20220119094503_populate_audit_event_streaming_verification_token.rb

+# frozen_string_literal: true
+
+class PopulateAuditEventStreamingVerificationToken < Gitlab::Database::Migration[1.0]
+  class ExternalAuditEventDestination < ActiveRecord::Base
+    self.table_name = 'audit_events_external_audit_event_destinations'
+
+    def regenerate_verification_token
+      update!(verification_token: SecureRandom.base58(24))
+    end
+  end
+
+  def up
+    ExternalAuditEventDestination.all.each { |destination| destination.regenerate_verification_token }
+  end
+
+  def down
+    # no-op
+  end
+end

db/schema_migrations/20220105082217

+448481ec9f7dd58d267e3660a49161c0e14baca35e640c59b27f2ebc4367b62a
\ No newline at end of file

db/schema_migrations/20220117082611

+28df9a8b5bf73bc33275cfe47f260788fa3263680a97128e086fd1698ccac1d8
\ No newline at end of file

db/schema_migrations/20220119094023

+4eddd356d87ce8fc8168dabe678211239e8d4051804d51d3bdce8cc137fa5a0d
\ No newline at end of file

db/schema_migrations/20220119094503

+1048b3a9744f212297c0a3aba176556e92e85f199ac861eb3ee4183eff002860
\ No newline at end of file

db/structure.sql

@@ -10775,7 +10775,9 @@ CREATE TABLE audit_events_external_audit_event_destinations (
     destination_url text NOT NULL,
     created_at timestamp with time zone NOT NULL,
     updated_at timestamp with time zone NOT NULL,
-    CONSTRAINT check_2feafb9daf CHECK ((char_length(destination_url) <= 255))
+    verification_token text,
+    CONSTRAINT check_2feafb9daf CHECK ((char_length(destination_url) <= 255)),
+    CONSTRAINT check_8ec80a7d06 CHECK ((char_length(verification_token) <= 24))
 );
 
 CREATE SEQUENCE audit_events_external_audit_event_destinations_id_seq
@@ -25355,6 +25357,8 @@ CREATE INDEX index_approvers_on_user_id ON approvers USING btree (user_id);
 
 CREATE UNIQUE INDEX index_atlassian_identities_on_extern_uid ON atlassian_identities USING btree (extern_uid);
 
+CREATE UNIQUE INDEX index_audit_events_external_audit_on_verification_token ON audit_events_external_audit_event_destinations USING btree (verification_token);
+
 CREATE INDEX index_authentication_events_on_provider ON authentication_events USING btree (provider);
 
 CREATE INDEX index_authentication_events_on_provider_user_id_created_at ON authentication_events USING btree (provider, user_id, created_at) WHERE (result = 1);

doc/administration/audit_event_streaming.md

@@ -13,7 +13,7 @@ FLAG:
 On self-managed GitLab, by default this feature is available. To hide the feature per group, ask an administrator to [disable the feature flag](../administration/feature_flags.md) named `ff_external_audit_events_namespace`. On GitLab.com, this feature is available.
 
 Event streaming allows owners of top-level groups to set an HTTP endpoint to receive **all** audit events about the group, and its
-subgroups and projects.
+subgroups and projects as structured JSON.
 
 Top-level group owners can manage their audit logs in third-party systems such as Splunk, using the Splunk
 [HTTP Event Collector](https://docs.splunk.com/Documentation/Splunk/8.2.2/Data/UsetheHTTPEventCollector). Any service that can receive
@@ -37,6 +37,7 @@ mutation {
     externalAuditEventDestination {
       destinationUrl
       group {
+      verificationToken
         name
       }
     }
@@ -60,6 +61,7 @@ query {
     externalAuditEventDestinations {
       nodes {
         destinationUrl
+        verificationToken
         id
       }
     }
@@ -68,3 +70,13 @@ query {
 ```
 
 If the resulting list is empty, then audit event streaming is not enabled for that group.
+
+## Verify event authenticity
+
+> [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/345424) in GitLab 14.8.
+
+Each streaming destination has a unique verification token (`verificationToken`) that can be used to verify the authenticity of the event. This
+token is generated when the event destination is created and cannot be changed.
+
+Each streamed event contains a random alphanumeric identifier for the `X-Gitlab-Event-Streaming-Token` HTTP header that can be verified against
+the destination's value when [listing streaming destinations](#list-currently-enabled-streaming-destinations).

doc/api/graphql/reference/index.md

@@ -10531,6 +10531,7 @@ Represents an external resource to send audit events to.
 | <a id="externalauditeventdestinationdestinationurl"></a>`destinationUrl` | [`String!`](#string) | External destination to send audit events to. |
 | <a id="externalauditeventdestinationgroup"></a>`group` | [`Group!`](#group) | Group the destination belongs to. |
 | <a id="externalauditeventdestinationid"></a>`id` | [`ID!`](#id) | ID of the destination. |
+| <a id="externalauditeventdestinationverificationtoken"></a>`verificationToken` | [`String!`](#string) | Verification token to validate source of event. |
 
 ### `ExternalIssue`
 

ee/app/graphql/ee/types/audit_events/external_audit_event_destination_type.rb

@@ -18,6 +18,10 @@ module EE
         field :group, ::Types::GroupType,
               null: false,
               description: 'Group the destination belongs to.'
+
+        field :verification_token, GraphQL::Types::String,
+              null: false,
+              description: 'Verification token to validate source of event.'
       end
     end
   end

ee/app/models/audit_events/external_audit_event_destination.rb

@@ -12,5 +12,6 @@ module AuditEvents
 
     validates :destination_url, public_url: true, presence: true
     validates :destination_url, uniqueness: { scope: :namespace_id }, length: { maximum: 255 }
+    has_secure_token :verification_token, length: 24
   end
 end

ee/app/workers/audit_events/audit_event_streaming_worker.rb

@@ -4,6 +4,7 @@ module AuditEvents
   class AuditEventStreamingWorker
     include ApplicationWorker
 
+    HEADER_KEY = "X-Gitlab-Event-Streaming-Token"
     REQUEST_BODY_SIZE_LIMIT = 25.megabytes
 
     # Audit Events contains a unique ID so the ingesting system should
@@ -26,7 +27,9 @@ module AuditEvents
 
       group.external_audit_event_destinations.each do |destination|
         Gitlab::HTTP.post(destination.destination_url,
-                          body: Gitlab::Json::LimitedEncoder.encode(audit_event.as_json, limit: REQUEST_BODY_SIZE_LIMIT), use_read_total_timeout: true)
+                          body: Gitlab::Json::LimitedEncoder.encode(audit_event.as_json, limit: REQUEST_BODY_SIZE_LIMIT),
+                          use_read_total_timeout: true,
+                          headers: { HEADER_KEY => destination.verification_token })
       end
     end
 

ee/spec/models/audit_events/external_audit_event_destination_spec.rb

@@ -13,6 +13,7 @@ RSpec.describe AuditEvents::ExternalAuditEventDestination do
     it { is_expected.to validate_uniqueness_of(:destination_url).scoped_to(:namespace_id) }
     it { is_expected.to validate_length_of(:destination_url).is_at_most(255) }
     it { is_expected.to validate_presence_of(:destination_url) }
+    it { is_expected.to have_db_column(:verification_token).of_type(:text) }
   end
 
   it_behaves_like 'includes Limitable concern' do

ee/spec/requests/api/graphql/group/external_audit_event_destinations_spec.rb

@@ -35,9 +35,11 @@ RSpec.describe 'getting a list of external audit event destinations for a group'
     it 'returns the groups external audit event destinations' do
       post_graphql(query, current_user: current_user)
 
+      verification_token_regex = /\A\w{24}\z/i
+
       expect(graphql_data_at(*path)).to contain_exactly(
-        a_hash_including('destinationUrl' => destination_1.destination_url),
-        a_hash_including('destinationUrl' => destination_2.destination_url)
+        a_hash_including('destinationUrl' => destination_1.destination_url, 'verificationToken' => a_string_matching(verification_token_regex)),
+        a_hash_including('destinationUrl' => destination_2.destination_url, 'verificationToken' => a_string_matching(verification_token_regex))
       )
     end
   end

ee/spec/workers/audit_events/audit_event_streaming_worker_spec.rb

@@ -30,6 +30,12 @@ RSpec.describe AuditEvents::AuditEventStreamingWorker do
 
         subject
       end
+
+      it 'sends the correct verification header' do
+        expect(Gitlab::HTTP).to receive(:post).with(an_instance_of(String), a_hash_including(headers: { 'X-Gitlab-Event-Streaming-Token' => anything })).once
+
+        subject
+      end
     end
 
     context 'when the group has several destinations' do

spec/migrations/populate_audit_event_streaming_verification_token_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe PopulateAuditEventStreamingVerificationToken do
+  let(:groups) { table(:namespaces) }
+  let(:destinations) { table(:audit_events_external_audit_event_destinations) }
+  let(:migration) { described_class.new }
+
+  let!(:group) { groups.create!(name: 'test-group', path: 'test-group') }
+  let!(:destination) { destinations.create!(namespace_id: group.id, destination_url: 'https://example.com/destination', verification_token: nil) }
+
+  describe '#up' do
+    it 'adds verification tokens to records created before the migration' do
+      expect do
+        migrate!
+        destination.reload
+      end.to change { destination.verification_token }.from(nil).to(a_string_matching(/\w{24}/))
+    end
+  end
+end