GitLab Event Collector

Analytics & SIEM · GitLab

Details

ID	`GitLab Event Collector`
Provider	GitLab Inc
Category	Analytics & SIEM
From Version	6.8.0
Docker Image	`demisto/fastapi:0.125.0.10158186`
Supported Modules	Agentix XSIAM

README

An event collector for GitLab audit events using Gitlab’s API.

Audit events API documentation

Prerequisites

To retrieve audit events using the API, you must authenticate yourself as an Administrator.

You must use Personal access tokens:

Create a Personal Access Token

In the upper-right corner, select your avatar.
Select Edit profile.
On the left sidebar, select Personal access tokens.
Select Add new token.
In Token name, enter a name for the token.
Optional. In Token description, enter a description for the token.
In Expiration date, enter an expiration date for the token.
- The token expires on that date at midnight UTC. A token with the expiration date of 2024-01-01 expires at 00:00:00 UTC on 2024-01-01.
- If you do not enter an expiration date, the expiration date is automatically set to 365 days later than the current date.
- By default, this date can be a maximum of 365 days later than the current date. In GitLab 17.6 or later, you can extend this limit to 400 days.
Select the desired scopes (see PAT scopes).
Select Create personal access token.

Configure Gitlab Event Collector in Cortex

Parameter	Description	Required
Server URL		True
API Key	The personal access token created above with Administrator authorization.	True
Fetch Instance Audit Events	When checked, the fetch mechanism will fetch events from the audit_events endpoint. That endpoint requires Administrator authorization. See Audit Events API documentation for more details.
Groups IDs		False
Projects IDS		False
First fetch timestamp (<number> <time unit>, for example, 12 hours, 7 days, 3 months, 1 year)		True
The maximum number of events per fetch for each event type	Each fetch will bring the `limit` number of events for each type (audits, groups and projects) and each group/project ID. For example, if `limit` is set to 500 and groups/projects IDs are given as well, then the fetch will bring 500 audit events and 500 group/project events for each group/project ID.	False
Trust any certificate (not secure)		False
Use system proxy settings		False

Commands

You can execute the following command from the Cortex XSOAR CLI, as part of an automation, or in a playbook. After you successfully execute a command, a DBot message appears in the War Room with the command details.

gitlab-get-events

Manual command to fetch events and display them.

Base Command

gitlab-get-events

Input

Argument Name	Description	Required
should_push_events	Set this argument to True in order to create events, otherwise the command will only display them. Default is False.	True

Context Output

There is no context output for this command.

Configuration parameters

url — Server URL (required)
api_key — (required)
fetch_instance_audit_events — Fetch Instance Audit Events
group_ids — Groups IDs
project_ids — Projects IDS
after — First fetch timestamp (<number> <time unit>, for example, 12 hours, 7 days, 3 months, 1 year) (required)
limit — The maximum number of events per fetch for each event type
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings

Commands (1)

gitlab-get-events
Manual command to fetch events and display them.

Source

GitLabEventCollector.py GitLabEventCollector_test.py GitLabEventCollector.yml GitLabEventCollector_description.md

Download History Raw

import pytest

from GitLabEventCollector import Client, prepare_query_params
from requests import Session


class MockResponse:
    def __init__(self, data: list):
        self.ok = True
        self.status_code = 200
        self.data = data
        self.links = None

    def json(self):
        return self.data

    def raise_for_status(self):
        pass


""" Test methods """


@pytest.mark.parametrize(
    "params, last_run, expected_params_url",
    [
        ({"after": "02/02/2022T15:00:00Z"}, {}, "pagination=keyset&created_after=2022-02-02T15:00:00Z&per_page=100"),
        (
            {"after": "02/02/2022T15:01:00Z"},
            {"next_url": "pagination=keyset&created_after=2022-02-02T15:00:00Z&per_page=100&cursor=examplecursor"},
            "pagination=keyset&created_after=2022-02-02T15:00:00Z&per_page=100&cursor=examplecursor",
        ),
    ],
)
def test_gitlab_events_params_good(params, last_run, expected_params_url):
    """
    Given:
        - Various params and LastRun dictionary values.
    When:
        - preparing the parameters.
    Then:
        - Make sure they are parsed correctly into the URL suffix.
    """
    assert expected_params_url == prepare_query_params(params, last_run)


def test_fetch_events(mocker):
    """
    Given:
        - fetch-events call, where last_id = 1 and fetch_instance_audit_events = True
    When:
        - Three following results are retrieved from the API:
            1. id = 1, created_at = '2023-10-28T20:29:34.872Z'
            2. id = 2, created_at = '2023-10-28T20:29:34.872Z'
            3. id = 3, created_at = '2023-10-28T20:29:34.872Z'
    Then:
        - Make sure only events 2 and 3 are returned (1 should not).
        - Verify the new lastRun is calculated correctly.
    """
    from GitLabEventCollector import fetch_events_command, main, demisto

    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(demisto, "params", return_value={"url": ""})
    mocker.patch.object(demisto, "getLastRun", return_value={"audit_events": {"last_id": "1"}})

    last_run = {"audit_events": {"last_id": "1"}}
    types = {"instance_events": True}
    mock_response = MockResponse(
        [
            {"id": "3", "created_at": "2023-10-28T20:29:34.872Z"},
            {"id": "2", "created_at": "2023-10-28T20:29:34.872Z"},
            {"id": "1", "created_at": "2023-10-28T20:29:34.872Z"},
        ]
    )
    mocker.patch.object(Session, "request", return_value=mock_response)
    events, _, new_last_run = fetch_events_command(Client(base_url=""), params={}, last_run=last_run, event_type_management=types)

    assert len(events) == 2
    assert events[0].get("id") != "1"
    assert new_last_run["audit_events"]["last_id"] == "3"

    # Tests main()
    mock_setLastRun = mocker.patch.object(demisto, "setLastRun")
    mock_events_result = mocker.patch("GitLabEventCollector.send_events_to_xsiam")
    main()

    assert len(mock_events_result.call_args[0][0]) == 2
    assert mock_events_result.call_args[0][0][0].get("id") != "1"
    assert mock_setLastRun.call_args[0][0]["audit_events"]["last_id"] == "3"


def test_fetch_events_with_two_iterations(mocker):
    """
    Given:
        - fetch-events command execution.
    When:
        - Limit parameter value is 300.
        - A single logs API call retrieves 200 events.
        - first_id is saved in lastRun.
    Then:
        - Make sure the logs API is called twice.
        - Make sure the first event has the same id as the first_id in the lastRun.
    """
    from GitLabEventCollector import fetch_events_command

    first_id = 2
    last_run = {"groups": {}, "projects": {}, "audit_events": {"first_id": first_id}}
    types = {"instance_events": True}
    params = {"limit": 300}

    mock_response = MockResponse([{"id": i, "created_at": 1521214343} for i in range(200)])
    mock_response.links = {"next": {"url": "https://example.com?param=value"}}
    mock_request = mocker.patch.object(Session, "request", return_value=mock_response)
    events, _, _ = fetch_events_command(Client(base_url=""), params=params, last_run=last_run, event_type_management=types)

    assert events[0].get("id") == first_id
    assert mock_request.call_count == 2


def test_fetch_events_with_groups_and_projects(mocker):
    """
    Given:
        - fetch-events command execution.
    When:
        - Limit parameter value is 300.
        - A single logs API call retrieves 200 events.
    Then:
        - Make sure the logs API is called twice.
    """
    from GitLabEventCollector import fetch_events_command

    last_run = {"groups": {}, "projects": {"last_id": "2"}, "audit_events": {"last_id": "1"}}

    mock_response = MockResponse(
        [
            {"id": "5", "created_at": 1521214345},
            {"id": "4", "created_at": 1521214343},
            {"id": "3", "created_at": 1521214345},
            {"id": "2", "created_at": 1521214343},
            {"id": "1", "created_at": 1521214343},
        ]
    )

    mocker.patch.object(Session, "request", return_value=mock_response)
    audit_events, group_and_project_events, new_last_run = fetch_events_command(
        Client(base_url=""),
        params={"limit": 4, "url": ""},
        last_run=last_run,
        event_type_management={"groups_ids": [1], "projects_ids": [2, 3, 4], "instance_events": True},
    )

    assert len(audit_events) == 4
    assert len(group_and_project_events) == 7
    assert new_last_run["audit_events"]["last_id"] == "5"
    assert new_last_run["projects"]["last_id"] == "5"
    assert new_last_run["groups"]["last_id"] == "5"
    assert new_last_run["groups"]["first_id"] == "1"
    assert "first_id" not in new_last_run["projects"]


def test_get_events(mocker):
    """
    Given:
        - gitlab-get-events call
    When:
        - Three following results are retrieved from the API:
            1. id = 1, created_at = 1521214343
            2. id = 2, created_at = 1521214343
            3. id = 3, created_at = 1521214345
    Then:
        - Make sure all of the events are returned as part of the CommandResult.
    """
    from GitLabEventCollector import get_events_command, main, demisto

    mocker.patch.object(demisto, "command", return_value="gitlab-get-events")
    mocker.patch.object(demisto, "params", return_value={"url": ""})

    mock_response = MockResponse(
        [
            {"id": "3", "created_at": 1521214345},
            {"id": "2", "created_at": 1521214343},
            {"id": "1", "created_at": 1521214343},
        ]
    )
    mocker.patch.object(Session, "request", return_value=mock_response)
    _, results = get_events_command(Client(base_url=""), args={})

    assert len(results.raw_response) == 3
    assert results.raw_response == mock_response.json()

    # Tests main()
    mock_results = mocker.patch.object(demisto, "results")
    main()
    assert mock_results.call_args[0][0]["Contents"] == mock_response.json()


def test_test_module(mocker):
    """
    Given:
        - test-module call.
    When:
        - A response with an OK status_code is retrieved from the API call.
    Then:
        - Make sure 'ok' is returned.
    """
    from GitLabEventCollector import test_module_command, main, demisto

    params = {"url": ""}
    mocker.patch.object(demisto, "command", return_value="test-module")
    mocker.patch.object(demisto, "params", return_value=params)

    mocker.patch.object(Session, "request", return_value=MockResponse([]))
    assert test_module_command(Client(base_url=""), {"url": ""}, {"groups_ids": [1, 2], "projects_ids": [3, 4, 5]}) == "ok"

    # Tests main()
    mock_results = mocker.patch.object(demisto, "results")
    main()
    assert mock_results.call_args[0][0] == "ok"