QRadarCreateAQLQuery

Build QRadar AQL Query.

Details

ID	`QRadarCreateAQLQuery`
Language	python
From Version	6.0.0
Docker Image	`demisto/python3:3.12.13.10116658`
Tags	Utility

README

Build QRadar AQL Query. (Available from Cortex XSOAR 6.0.0).

Script Data

Name	Description
Script Type	python3
Tags	Utility
Cortex XSOAR Version	6.0.0

Inputs

Argument Name	Description
base_values_to_search	The values of the first field to search. This can be a single value or a comma-separated list of values. For example admin1,admin2.
base_fields_to_search	The field names of the first field to search. This can be a single value or a comma-separated list of values. For example admin1,admin2.
base_field_state	The state of the second field to search, meaning whether the values in the field should be included or excluded. Valid options are include or exclude.
base_field_match	Whether the values of the second field should be an exact match or a partial match. Valid options are exact or partial.
select_fields	The list of fields to select within the AQL query. The default fields are DATEFORMAT(devicetime,’dd-MM-yyyy hh:mm’),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username
time_frame	Time frame as used in AQL examples can be LAST 7 DAYS START ‘2019-09-25 15:51’ STOP ‘2019-09-25 17:51’. For more examples, view IBM’s AQL documentation.
first_additional_values	The values of the second field to search. This can be a single value or a comma-separated list of values. For example admin1,admin2.
first_additional_fields	The field names of the second field to search. This can be a single value or a comma-separated list of values. For example admin1,admin2.
first_additional_field_state	The state of the second field to search, meaning whether the values in the field should be included or excluded. Valid options are include or exclude.
first_additional_field_match	Whether the values of the second field should be an exact match or a partial match. Valid options are exact or partial.
second_additional_values	The values of the third field to search. This can be a single value or a comma-separated list of values. For example admin1,admin2
second_additional_fields	The field names of the third field to search. This can be a single value or a comma-separated list of values. For example username,user
second_additional_field_state	The state of the third field to search, meaning whether the values in the field should be included or excluded. Valid options are include or exclude.
second_additional_field_match	Whether the values of the third field should be an exact match or a partial match. Valid options are exact or partial. When choosing exact, the AQL query will use the = operator. When choosing partial, the AQL query will use ILIKE and add ‘%%’ to the values.

Outputs

Path	Description	Type
QRadarQuery	The resultant AQL query based on the inputs.	string

Script Example Search for a hash where we dont know the field

!QRadarCreateAQLQuery base_field_match=partial base_values_to_search=2367666DB8DFF58982A74695760E3EF0ACEBD050

Context Example

{
    "QRadarQuery": "select DATEFORMAT(devicetime,'dd-MM-yyyy hh:mm'),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (UTF8(payload) ILIKE '%2367666DB8DFF58982A74695760E3EF0ACEBD050%') LAST 1 HOURS"
}

Human Readable Output

select DATEFORMAT(devicetime,’dd-MM-yyyy hh:mm’),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (UTF8(payload) ILIKE ‘%2367666DB8DFF58982A74695760E3EF0ACEBD050%’) LAST 1 HOURS

Script Example Search for a hash in specific fields

!QRadarCreateAQLQuery base_field_match=exact base_values_to_search=2367666DB8DFF58982A74695760E3EF0ACEBD050 base_fields_to_search=sha1,sha1-hash

Context Example

{
    "QRadarQuery": "select DATEFORMAT(devicetime,'dd-MM-yyyy hh:mm'),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = '2367666DB8DFF58982A74695760E3EF0ACEBD050' OR sha1-hash = '2367666DB8DFF58982A74695760E3EF0ACEBD050') LAST 1 HOURS"
}

Human Readable Output

select DATEFORMAT(devicetime,’dd-MM-yyyy hh:mm’),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = ‘2367666DB8DFF58982A74695760E3EF0ACEBD050’ OR sha1-hash = ‘2367666DB8DFF58982A74695760E3EF0ACEBD050’) LAST 1 HOURS

Script Example Search for user and hash

!QRadarCreateAQLQuery base_field_match=exact base_values_to_search=2367666DB8DFF58982A74695760E3EF0ACEBD050 base_fields_to_search=sha1,sha1-hash first_additional_fields=username first_additional_values=admin

Context Example

{
    "QRadarQuery": "select DATEFORMAT(devicetime,'dd-MM-yyyy hh:mm'),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = '2367666DB8DFF58982A74695760E3EF0ACEBD050' OR sha1-hash = '2367666DB8DFF58982A74695760E3EF0ACEBD050') AND (username = 'admin') LAST 1 HOURS"
}

Human Readable Output

select DATEFORMAT(devicetime,’dd-MM-yyyy hh:mm’),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = ‘2367666DB8DFF58982A74695760E3EF0ACEBD050’ OR sha1-hash = ‘2367666DB8DFF58982A74695760E3EF0ACEBD050’) AND (username = ‘admin’) LAST 1 HOURS

Script Example Search for a hash in payload that doesnt contain admin

!QRadarCreateAQLQuery base_field_match=exact base_values_to_search=2367666DB8DFF58982A74695760E3EF0ACEBD050 base_fields_to_search=sha1 first_additional_field_state=exclude first_additional_field_match=partial first_additional_values=admin

Context Example

{
    "QRadarQuery": "select DATEFORMAT(devicetime,'dd-MM-yyyy hh:mm'),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = '2367666DB8DFF58982A74695760E3EF0ACEBD050') AND (UTF8(payload) NOT ILIKE '%admin%') LAST 1 HOURS"
}

Human Readable Output

select DATEFORMAT(devicetime,’dd-MM-yyyy hh:mm’),LOGSOURCENAME(logsourceid),CATEGORYNAME(category),QIDNAME(qid),sourceip,destinationip,username from events where (sha1 = ‘2367666DB8DFF58982A74695760E3EF0ACEBD050’) AND (UTF8(payload) NOT ILIKE ‘%admin%’) LAST 1 HOURS

Source

QRadarCreateAQLQuery.py QRadarCreateAQLQuery_test.py QRadarCreateAQLQuery.yml

Download History Raw

import pytest
from CommonServerPython import *
from QRadarCreateAQLQuery import *


class TestFieldsSection:
    def test_sanity(self):
        assert fields_section(["field"], ["value"]) == "(field = 'value')"

    def test_with_space(self):
        assert fields_section(["field name"], ["value"]) == "('field name' = 'value')"

    def test_multiple_fields(self):
        assert fields_section(["field1", "field2"], ["value"]) == "(field1 = 'value' OR field2 = 'value')"

    def test_multiple_values(self):
        assert fields_section(["field"], ["value1", "value2"]) == "(field = 'value1' OR field = 'value2')"

    def test_multiple_fields_and_values(self):
        assert (
            fields_section(["field1", "field2"], ["value1", "value2"])
            == "(field1 = 'value1' OR field1 = 'value2' OR field2 = 'value1' OR field2 = 'value2')"
        )

    def test_not_equal(self):
        assert fields_section(["field"], ["value"], match_rule=MatchRule.NOT_EQUAL) == "(field != 'value')"

    def test_ilike(self):
        assert fields_section(["field"], ["value"], match_rule=MatchRule.ILIKE) == "(field ILIKE '%value%')"

    def test_not_not_ilike(self):
        assert fields_section(["field"], ["value"], match_rule=MatchRule.NOT_ILIKE) == "(field NOT ILIKE '%value%')"


def test_complete_query():
    assert (
        complete_query("select_fields", "combined_sections", "time_frame")
        == "select select_fields from events where combined_sections time_frame"
    )


class TestPrepareArgs:
    def test_with_empty(self):
        assert prepare_args({"some emty ergument": ""}) == {}

    def test_with_base_args(self):
        test = "test"
        original_args = {
            "base_values_to_search": test,
            "base_fields_to_search": test,
            "base_field_state": test,
            "base_field_match": test,
        }
        expected_args = {
            "base_additional_values": test,
            "base_additional_fields": test,
            "base_additional_field_state": test,
            "base_additional_field_match": test,
        }
        assert prepare_args(original_args) == expected_args

    def test_base_kyas_not_found(self):
        assert prepare_args({}) == {}


def test_original_key_name():
    assert original_key_name("non_existing_key") == "non_existing_key"
    assert original_key_name("base_additional_values") == "base_values_to_search"
    assert original_key_name("base_additional_fields") == "base_fields_to_search"
    assert original_key_name("base_additional_field_state") == "base_field_state"
    assert original_key_name("base_additional_field_match") == "base_field_match"


class TestCreateSectionsStr:
    def test_failure_with_non_existing_base(self):
        with pytest.raises(DemistoException, match="base arguments not given correctly"):
            create_sections_str({})

    def test_go_over_all_3_arg_groups(self, mocker):
        mocker.patch("QRadarCreateAQLQuery.prepare_section", return_value={})
        fields_section_mocker = mocker.patch("QRadarCreateAQLQuery.fields_section", return_value="")
        create_sections_str({})
        assert fields_section_mocker.call_count == 3

    def test_operator_concat(self, mocker):
        mocker.patch("QRadarCreateAQLQuery.prepare_section", return_value={})
        mocker.patch("QRadarCreateAQLQuery.fields_section", return_value="test")
        assert create_sections_str({}) == "test AND test AND test"


class TestPrepareSection:
    def test_without_the_section_args(self):
        with pytest.raises(SectionNotFound, match="section name"):
            prepare_section({}, "section name")

    def test_singel_value_and_field(self):
        input_args = {
            "_additional_values": "test_val",
            "_additional_fields": "test_field",
            "_additional_field_state": "include",
            "_additional_field_match": "exact",
        }
        output_args = prepare_section(input_args, "")
        assert output_args["values_list"] == ["test_val"]
        assert output_args["fields_list"] == ["test_field"]

    def test_values_list(self):
        input_args = {
            "_additional_values": "test_val1,test_val2",
            "_additional_fields": "test_field1,test_field2",
            "_additional_field_state": "include",
            "_additional_field_match": "exact",
        }
        output_args = prepare_section(input_args, "")
        assert output_args["values_list"] == ["test_val1", "test_val2"]
        assert output_args["fields_list"] == ["test_field1", "test_field2"]

    class TestWithoutFields:
        def test_with_exact_field_match(self):
            args = {"_additional_values": "test", "_additional_field_state": "include", "_additional_field_match": "exact"}
            with pytest.raises(KeyError, match="_additional_fields"):
                prepare_section(args, "")

        def test_with_partial_field_match(self):
            input_args = {
                "_additional_values": "test_val1,test_val2",
                "_additional_field_state": "include",
                "_additional_field_match": "partial",
            }
            output_args = prepare_section(input_args, "")
            assert output_args["fields_list"] == ["UTF8(payload)"]

    class TestMachRules:
        @classmethod
        def setup_method(cls):
            cls._args = {"_additional_values": "test_val", "_additional_fields": "test_field"}

        def test_match_rule_equal(self):
            self._args.update({"_additional_field_state": "include", "_additional_field_match": "exact"})
            assert prepare_section(self._args, "")["match_rule"] == MatchRule.EQUAL

        def test_match_rule_not_equal(self):
            self._args.update({"_additional_field_state": "exclude", "_additional_field_match": "exact"})
            assert prepare_section(self._args, "")["match_rule"] == MatchRule.NOT_EQUAL

        def test_match_rule_ilike(self):
            self._args.update({"_additional_field_state": "include", "_additional_field_match": "partial"})
            assert prepare_section(self._args, "")["match_rule"] == MatchRule.ILIKE

        def test_match_rule_not_ilike(self):
            self._args.update({"_additional_field_state": "exclude", "_additional_field_match": "partial"})
            assert prepare_section(self._args, "")["match_rule"] == MatchRule.NOT_ILIKE


class TestKeyErrors:
    @staticmethod
    def mocks(args, mocker):
        mocker.patch.object(demisto, "args", return_value=args)
        return mocker.patch("QRadarCreateAQLQuery.return_error")

    def test_with_required_key(self, mocker):
        mock_return_error = self.mocks({}, mocker)
        main()
        mock_return_error.assert_called_once_with("Missing time_frame.")

    def test_with_base_key(self, mocker):
        args = {
            "time_frame": "test",
            "select_fields": "test",
            "base_values_to_search": "test",
            "base_field_state": "exclude",
            "base_field_match": "exact",
        }
        mock_return_error = self.mocks(args, mocker)
        main()
        mock_return_error.assert_called_once_with("Missing base_fields_to_search.")

    def test_with_other_key(self, mocker):
        args = {
            "time_frame": "test",
            "select_fields": "test",
            "base_values_to_search": "test",
            "base_field_state": "exclude",
            "base_field_match": "partial",
            "first_additional_values": "test",
            "first_additional_field_state": "exclude",
            "first_additional_field_match": "exact",
        }
        mock_return_error = self.mocks(args, mocker)
        main()
        mock_return_error.assert_called_once_with("Missing first_additional_fields.")


def test_main_general_exception(mocker):
    error = Exception("test")

    def raising():
        raise error

    mocker.patch.object(demisto, "args", new=raising)
    mock_return_error = mocker.patch("QRadarCreateAQLQuery.return_error")
    main()
    mock_return_error.assert_called_once_with("test", error)