Google cloud logging streaming strategy

This commit adds streaming strategy to allow
streaming of audit events to google cloud
logging service

EE: true
Changelog: added

ee/app/models/audit_events/google_cloud_logging_configuration.rb

@@ -42,6 +42,14 @@ class GoogleCloudLoggingConfiguration < ApplicationRecord
 
     validate :root_level_group?
 
+    def allowed_to_stream?(*)
+      true
+    end
+
+    def full_log_path
+      "projects/#{google_project_id_name}/logs/#{log_id_name}"
+    end
+
     private
 
     def root_level_group?

ee/lib/audit_events/external_destination_streamer.rb

@@ -6,7 +6,8 @@ class ExternalDestinationStreamer
 
     STRATEGIES = [
       AuditEvents::Strategies::GroupExternalDestinationStrategy,
-      AuditEvents::Strategies::InstanceExternalDestinationStrategy
+      AuditEvents::Strategies::InstanceExternalDestinationStrategy,
+      AuditEvents::Strategies::GoogleCloudLoggingDestinationStrategy
     ].freeze
 
     def initialize(event_name, audit_event)

ee/lib/audit_events/strategies/google_cloud_logging_destination_strategy.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Strategies
+    class GoogleCloudLoggingDestinationStrategy < ExternalDestinationStrategy
+      def initialize(audit_operation, audit_event)
+        @logger = GoogleCloud::LoggingService::Logger.new
+
+        super(audit_operation, audit_event)
+      end
+
+      def streamable?
+        group = audit_event.root_group_entity
+        return false if group.nil?
+        return false unless group.licensed_feature_available?(:external_audit_events)
+
+        group.google_cloud_logging_configurations.exists?
+      end
+
+      private
+
+      def destinations
+        group = audit_event.root_group_entity
+        group.present? ? group.google_cloud_logging_configurations.to_a : []
+      end
+
+      def track_and_stream(destination)
+        track_audit_event_count
+
+        @logger.log(destination.client_email, destination.private_key, json_payload(destination))
+      end
+
+      def json_payload(destination)
+        { 'entries' => [log_entry(destination)] }.to_json
+      end
+
+      def log_entry(destination)
+        {
+          'logName' => destination.full_log_path,
+          'resource' => {
+            'type' => 'global'
+          },
+          'severity' => 'INFO',
+          'jsonPayload' => ::Gitlab::Json.parse(request_body)
+        }
+      end
+    end
+  end
+end

ee/spec/lib/audit_events/external_destination_streamer_spec.rb

@@ -5,6 +5,10 @@
 RSpec.describe AuditEvents::ExternalDestinationStreamer, feature_category: :audit_events do
   before do
     stub_licensed_features(external_audit_events: true)
+
+    allow_next_instance_of(::GoogleCloud::Authentication) do |instance|
+      allow(instance).to receive(:generate_access_token).and_return("sample-token")
+    end
   end
 
   describe '#stream_to_destinations' do
@@ -25,10 +29,11 @@
       before do
         create(:external_audit_event_destination, group: group)
         create_list(:instance_external_audit_event_destination, 2)
+        create(:google_cloud_logging_configuration, group: group)
       end
 
-      it 'makes two HTTP calls' do
-        expect(Gitlab::HTTP).to receive(:post).thrice
+      it 'makes correct number of HTTP calls' do
+        expect(Gitlab::HTTP).to receive(:post).exactly(4).times
 
         subject
       end
@@ -45,27 +50,36 @@
       it { is_expected.to be_falsey }
     end
 
+    context 'when all of them are streamable' do
+      before do
+        create(:external_audit_event_destination, group: group)
+        create(:instance_external_audit_event_destination)
+        create(:google_cloud_logging_configuration, group: group)
+      end
+
+      it { is_expected.to be_truthy }
+    end
+
     context 'when atleast one of them is streamable' do
-      context 'when all of them are streamable' do
+      context 'when only group external destination is streamable' do
         before do
           create(:external_audit_event_destination, group: group)
-          create(:instance_external_audit_event_destination)
         end
 
         it { is_expected.to be_truthy }
       end
 
-      context 'when group is streamable but instance is not' do
+      context 'when only instance destination is streamable' do
         before do
-          create(:external_audit_event_destination, group: group)
+          create(:instance_external_audit_event_destination)
         end
 
         it { is_expected.to be_truthy }
       end
 
-      context 'when instance is streamable but group is not' do
+      context 'when only google cloud logging destination is streamable' do
         before do
-          create(:instance_external_audit_event_destination)
+          create(:google_cloud_logging_configuration, group: group)
         end
 
         it { is_expected.to be_truthy }

ee/spec/lib/audit_events/strategies/google_cloud_logging_destination_strategy_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuditEvents::Strategies::GoogleCloudLoggingDestinationStrategy, feature_category: :audit_events do
+  let(:group) { build(:group) }
+  let(:event) { build(:audit_event, :group_event, target_group: group) }
+
+  let_it_be(:event_type) { 'audit_operation' }
+  let_it_be(:request_body) { { key: "value" }.to_json }
+
+  describe '#streamable?' do
+    subject { described_class.new(event_type, event).streamable? }
+
+    context 'when feature is not licensed' do
+      it { is_expected.to be_falsey }
+    end
+
+    context 'when feature is licensed' do
+      before do
+        stub_licensed_features(external_audit_events: true)
+      end
+
+      context 'when event group is nil' do
+        let_it_be(:event) { build(:audit_event) }
+
+        it { is_expected.to be_falsey }
+      end
+
+      context 'when group google cloud logging configurations does not exist' do
+        it { is_expected.to be_falsey }
+      end
+
+      context 'when group google cloud logging configurations exist' do
+        before do
+          create(:google_cloud_logging_configuration, group: group)
+        end
+
+        it { is_expected.to be_truthy }
+      end
+    end
+  end
+
+  describe '#destinations' do
+    subject { described_class.new(event_type, event).send(:destinations) }
+
+    context 'when event group is nil' do
+      let_it_be(:event) { build(:audit_event) }
+
+      it 'returns empty array' do
+        expect(subject).to eq([])
+      end
+    end
+
+    context 'when group google cloud logging configurations exist' do
+      it 'returns all the destinations' do
+        destination1 = create(:google_cloud_logging_configuration, group: group)
+        destination2 = create(:google_cloud_logging_configuration, group: group)
+
+        expect(subject).to match_array([destination1, destination2])
+      end
+    end
+  end
+
+  describe '#track_and_stream' do
+    let(:instance) { described_class.new(event_type, event) }
+    let!(:destination) { create(:google_cloud_logging_configuration, group: group) }
+
+    subject(:track_and_stream) { instance.send(:track_and_stream, destination) }
+
+    context 'when a google cloud logging configuration exists' do
+      let(:expected_log_entry) do
+        [{ entries: {
+          'logName' => destination.full_log_path,
+          'resource' => {
+            'type' => 'global'
+          },
+          'severity' => 'INFO',
+          'jsonPayload' => ::Gitlab::Json.parse(request_body)
+        } }.to_json]
+      end
+
+      before do
+        allow_next_instance_of(GoogleCloud::LoggingService::Logger) do |instance|
+          allow(instance).to receive(:log).and_return(nil)
+        end
+        allow(instance).to receive(:request_body).and_return(request_body)
+      end
+
+      it 'tracks audit event count and calls logger' do
+        expect(instance).to receive(:track_audit_event_count)
+
+        allow_next_instance_of(GoogleCloud::LoggingService::Logger) do |logger|
+          expect(logger).to receive(:log).with(destination.client_email, destination.private_key, expected_log_entry)
+        end
+
+        track_and_stream
+      end
+    end
+  end
+end

ee/spec/models/audit_events/google_cloud_logging_configuration_spec.rb

@@ -84,6 +84,21 @@
     end
   end
 
+  describe '#allowed_to_stream?' do
+    it 'always returns true' do
+      expect(google_cloud_logging_config.allowed_to_stream?).to eq(true)
+    end
+  end
+
+  describe '#full_log_path' do
+    it 'returns the full log path for the google project' do
+      google_cloud_logging_config.google_project_id_name = "test-project"
+      google_cloud_logging_config.log_id_name = "test-log"
+
+      expect(google_cloud_logging_config.full_log_path).to eq("projects/test-project/logs/test-log")
+    end
+  end
+
   it_behaves_like 'includes Limitable concern' do
     subject { build(:google_cloud_logging_configuration, group: create(:group)) }
   end

lib/google_cloud/authentication.rb

+# frozen_string_literal: true
+
+module GoogleCloud
+  class Authentication
+    def initialize(scope:)
+      @scope = scope
+    end
+
+    def generate_access_token(client_email, private_key)
+      credentials = Google::Auth::ServiceAccountCredentials.make_creds(
+        json_key_io: StringIO.new({ client_email: client_email, private_key: private_key }.to_json),
+        scope: @scope
+      )
+      credentials.fetch_access_token!["access_token"]
+    rescue StandardError => e
+      ::Gitlab::ErrorTracking.track_exception(e, client_email: client_email)
+      nil
+    end
+  end
+end

lib/google_cloud/logging_service/logger.rb

+# frozen_string_literal: true
+
+module GoogleCloud
+  module LoggingService
+    class Logger
+      WRITE_URL = "https://logging.googleapis.com/v2/entries:write"
+      SCOPE = "https://www.googleapis.com/auth/logging.write"
+
+      def initialize
+        @auth = GoogleCloud::Authentication.new(scope: SCOPE)
+      end
+
+      def log(client_email, private_key, payload)
+        access_token = @auth.generate_access_token(client_email, private_key)
+
+        return unless access_token
+
+        headers = build_headers(access_token)
+
+        post(WRITE_URL, body: payload, headers: headers)
+      end
+
+      private
+
+      def build_headers(access_token)
+        { 'Authorization' => "Bearer #{access_token}", 'Content-Type' => 'application/json' }
+      end
+
+      def post(url, body:, headers:)
+        Gitlab::HTTP.post(
+          url,
+          body: body,
+          headers: headers
+        )
+      rescue URI::InvalidURIError => e
+        Gitlab::ErrorTracking.log_exception(e)
+      rescue *Gitlab::HTTP::HTTP_ERRORS
+      end
+    end
+  end
+end

spec/lib/google_cloud/authentication_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe GoogleCloud::Authentication, feature_category: :audit_events do
+  describe '#generate_access_token' do
+    let_it_be(:client_email) { 'test@example.com' }
+    let_it_be(:private_key) { 'private_key' }
+    let_it_be(:scope) { 'https://www.googleapis.com/auth/logging.write' }
+    let_it_be(:json_key_io) { StringIO.new({ client_email: client_email, private_key: private_key }.to_json) }
+
+    let(:service_account_credentials) { instance_double('Google::Auth::ServiceAccountCredentials') }
+
+    subject(:generate_access_token) do
+      described_class.new(scope: scope).generate_access_token(client_email, private_key)
+    end
+
+    before do
+      allow(Google::Auth::ServiceAccountCredentials).to receive(:make_creds).with(json_key_io: json_key_io,
+        scope: scope).and_return(service_account_credentials)
+      allow(StringIO).to receive(:new).with({ client_email: client_email,
+                                              private_key: private_key }.to_json).and_return(json_key_io)
+    end
+
+    context 'when credentials are valid' do
+      before do
+        allow(service_account_credentials).to receive(:fetch_access_token!).and_return({ 'access_token' => 'token' })
+      end
+
+      it 'calls make_creds with correct parameters' do
+        expect(Google::Auth::ServiceAccountCredentials).to receive(:make_creds).with(json_key_io: json_key_io,
+          scope: scope)
+
+        generate_access_token
+      end
+
+      it 'fetches access token' do
+        expect(generate_access_token).to eq('token')
+      end
+    end
+
+    context 'when an error occurs' do
+      before do
+        allow(service_account_credentials).to receive(:fetch_access_token!).and_raise(StandardError)
+      end
+
+      it 'handles the exception and returns nil' do
+        expect(Gitlab::ErrorTracking).to receive(:track_exception)
+        expect(generate_access_token).to be_nil
+      end
+    end
+  end
+end

spec/lib/google_cloud/logging_service/logger_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe GoogleCloud::LoggingService::Logger, feature_category: :audit_events do
+  let_it_be(:client_email) { 'test@example.com' }
+  let_it_be(:private_key) { 'private_key' }
+  let_it_be(:payload) { [{ logName: 'test-log' }.to_json] }
+  let_it_be(:access_token) { 'access_token' }
+  let_it_be(:expected_headers) do
+    { 'Authorization' => "Bearer #{access_token}", 'Content-Type' => 'application/json' }
+  end
+
+  subject(:log) { described_class.new.log(client_email, private_key, payload) }
+
+  describe '#log' do
+    context 'when access token is available' do
+      before do
+        allow_next_instance_of(GoogleCloud::Authentication) do |instance|
+          allow(instance).to receive(:generate_access_token).with(client_email, private_key).and_return(access_token)
+        end
+      end
+
+      it 'generates access token and calls Gitlab::HTTP.post with correct parameters' do
+        expect(Gitlab::HTTP).to receive(:post).with(
+          described_class::WRITE_URL,
+          body: payload,
+          headers: expected_headers
+        )
+
+        log
+      end
+
+      context 'when URI::InvalidURIError is raised' do
+        before do
+          allow(Gitlab::HTTP).to receive(:post).and_raise(URI::InvalidURIError)
+        end
+
+        it 'logs the exception' do
+          expect(Gitlab::ErrorTracking).to receive(:log_exception)
+
+          log
+        end
+      end
+    end
+
+    context 'when access token is not available' do
+      let(:access_token) { nil }
+
+      it 'does not call Gitlab::HTTP.post' do
+        allow_next_instance_of(GoogleCloud::Authentication) do |instance|
+          allow(instance).to receive(:generate_access_token).with(client_email, private_key).and_return(access_token)
+        end
+
+        expect(Gitlab::HTTP).not_to receive(:post)
+
+        log
+      end
+    end
+  end
+end