test_schemas.py

# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License
# 2.0; you may not use this file except in compliance with the Elastic License
# 2.0.

"""Test stack versioned schemas."""

import copy
import os
import unittest
import unittest.mock
import uuid
from pathlib import Path
from typing import Any

import eql
import pytest
from marshmallow import ValidationError
from semver import Version

from detection_rules import utils
from detection_rules.config import load_current_package_version
from detection_rules.esql_errors import EsqlSemanticError
from detection_rules.rule import TOMLRuleContents
from detection_rules.rule_loader import RuleCollection
from detection_rules.schemas import RULES_CONFIG, downgrade
from detection_rules.version_lock import VersionLockFile


class TestSchemas(unittest.TestCase):
    """Test schemas and downgrade functions."""

    @classmethod
    def setUpClass(cls):
        cls.current_version = load_current_package_version()

        # expected contents for a downgraded rule
        cls.v78_kql = {
            "description": "test description",
            "index": ["filebeat-*"],
            "language": "kuery",
            "name": "test rule",
            "query": "process.name:test.query",
            "risk_score": 21,
            "rule_id": str(uuid.uuid4()),
            "severity": "low",
            "type": "query",
            "threat": [
                {
                    "framework": "MITRE ATT&CK",
                    "tactic": {
                        "id": "TA0001",
                        "name": "Execution",
                        "reference": "https://attack.mitre.org/tactics/TA0001/",
                    },
                    "technique": [
                        {
                            "id": "T1059",
                            "name": "Command and Scripting Interpreter",
                            "reference": "https://attack.mitre.org/techniques/T1059/",
                        }
                    ],
                }
            ],
        }
        cls.v79_kql = dict(cls.v78_kql, author=["Elastic"], license="Elastic License v2")
        cls.v711_kql = copy.deepcopy(cls.v79_kql)
        # noinspection PyTypeChecker
        cls.v711_kql["threat"][0]["technique"][0]["subtechnique"] = [
            {"id": "T1059.001", "name": "PowerShell", "reference": "https://attack.mitre.org/techniques/T1059/001/"}
        ]
        # noinspection PyTypeChecker
        cls.v711_kql["threat"].append(
            {
                "framework": "MITRE ATT&CK",
                "tactic": {
                    "id": "TA0008",
                    "name": "Lateral Movement",
                    "reference": "https://attack.mitre.org/tactics/TA0008/",
                },
            }
        )

        cls.v79_threshold_contents = {
            "author": ["Elastic"],
            "description": "test description",
            "language": "kuery",
            "license": "Elastic License v2",
            "name": "test rule",
            "query": "process.name:test.query",
            "risk_score": 21,
            "rule_id": str(uuid.uuid4()),
            "severity": "low",
            "threshold": {
                "field": "destination.bytes",
                "value": 75,
            },
            "type": "threshold",
        }
        cls.v712_threshold_rule = dict(
            copy.deepcopy(cls.v79_threshold_contents),
            threshold={
                "field": ["destination.bytes", "process.args"],
                "value": 75,
                "cardinality": [{"field": "user.name", "value": 2}],
            },
        )

    def test_query_downgrade_7_x(self):
        """Downgrade a standard KQL rule."""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 7:
            return

        self.assertDictEqual(downgrade(self.v711_kql, "7.11"), self.v711_kql)
        self.assertDictEqual(downgrade(self.v711_kql, "7.9"), self.v79_kql)
        self.assertDictEqual(downgrade(self.v711_kql, "7.9.2"), self.v79_kql)
        self.assertDictEqual(downgrade(self.v711_kql, "7.8.1"), self.v78_kql)
        self.assertDictEqual(downgrade(self.v79_kql, "7.8"), self.v78_kql)
        self.assertDictEqual(downgrade(self.v79_kql, "7.8"), self.v78_kql)

        with self.assertRaises(ValueError):
            downgrade(self.v711_kql, "7.7")

        with self.assertRaises(ValueError):
            downgrade(self.v79_kql, "7.7")

        with self.assertRaises(ValueError):
            downgrade(self.v78_kql, "7.7", current_version="7.8")

    def test_versioned_downgrade_7_x(self):
        """Downgrade a KQL rule with version information"""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 7:
            return

        api_contents = self.v79_kql
        self.assertDictEqual(downgrade(api_contents, "7.9"), api_contents)
        self.assertDictEqual(downgrade(api_contents, "7.9.2"), api_contents)

        api_contents78 = api_contents.copy()
        api_contents78.pop("author")
        api_contents78.pop("license")

        self.assertDictEqual(downgrade(api_contents, "7.8"), api_contents78)

        with self.assertRaises(ValueError):
            downgrade(api_contents, "7.7")

    def test_threshold_downgrade_7_x(self):
        """Downgrade a threshold rule that was first introduced in 7.9."""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 7:
            return

        api_contents = self.v712_threshold_rule
        self.assertDictEqual(downgrade(api_contents, "7.13"), api_contents)
        self.assertDictEqual(downgrade(api_contents, "7.13.1"), api_contents)

        exc_msg = "Cannot downgrade a threshold rule that has multiple threshold fields defined"
        with self.assertRaisesRegex(ValueError, exc_msg):
            downgrade(api_contents, "7.9")

        v712_threshold_contents_single_field = copy.deepcopy(api_contents)
        v712_threshold_contents_single_field["threshold"]["field"].pop()

        with self.assertRaisesRegex(ValueError, "Cannot downgrade a threshold rule that has a defined cardinality"):
            downgrade(v712_threshold_contents_single_field, "7.9")

        v712_no_cardinality = copy.deepcopy(v712_threshold_contents_single_field)
        v712_no_cardinality["threshold"].pop("cardinality")
        self.assertEqual(downgrade(v712_no_cardinality, "7.9"), self.v79_threshold_contents)

        with self.assertRaises(ValueError):
            downgrade(v712_no_cardinality, "7.7")

        with self.assertRaisesRegex(ValueError, "Unsupported rule type"):
            downgrade(v712_no_cardinality, "7.8")

    def test_query_downgrade_8_x(self):
        """Downgrade a standard KQL rule."""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 8:
            return

    def test_versioned_downgrade_8_x(self):
        """Downgrade a KQL rule with version information"""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 8:
            return

    def test_threshold_downgrade_8_x(self):
        """Downgrade a threshold rule that was first introduced in 7.9."""
        if Version.parse(self.current_version, optional_minor_and_patch=True).major > 7:
            return

    def test_eql_validation(self):
        base_fields = {
            "author": ["Elastic"],
            "description": "test description",
            "index": ["filebeat-*"],
            "language": "eql",
            "license": "Elastic License v2",
            "name": "test rule",
            "risk_score": 21,
            "rule_id": str(uuid.uuid4()),
            "severity": "low",
            "type": "eql",
        }

        def build_rule(query):
            metadata = {
                "creation_date": "1970/01/01",
                "updated_date": "1970/01/01",
                "min_stack_version": load_current_package_version(),
            }
            data = base_fields.copy()
            data["query"] = query
            obj = {"metadata": metadata, "rule": data}
            return TOMLRuleContents.from_dict(obj)

        build_rule("""
            process where process.name == "cmd.exe"
        """)

        example_text_fields = [
            "client.as.organization.name.text",
            "client.user.full_name.text",
            "client.user.name.text",
            "destination.as.organization.name.text",
            "destination.user.full_name.text",
            "destination.user.name.text",
            "error.message",
            "error.stack_trace.text",
            "file.path.text",
            "file.target_path.text",
            "host.os.full.text",
            "host.os.name.text",
            "host.user.full_name.text",
            "host.user.name.text",
        ]
        for text_field in example_text_fields:
            with self.assertRaises(eql.parser.EqlSchemaError):
                build_rule(f"""
                        any where {text_field} == "some string field"
                """)

        with self.assertRaises(eql.EqlSyntaxError):
            build_rule("""
                    process where process.name == this!is$not#v@lid
            """)

        with self.assertRaises(eql.EqlSemanticError):
            build_rule("""
                    process where process.invalid_field == "hello world"
            """)

        with self.assertRaises(eql.EqlTypeMismatchError):
            build_rule("""
                    process where process.pid == "some string field"
            """)

    def test_response_actions_validation(self) -> None:
        """Test that response actions are properly validated in the schema."""
        base_fields: dict[str, Any] = {
            "author": ["Elastic"],
            "description": "test description",
            "index": ["logs-endpoint.events.*"],
            "language": "kuery",
            "license": "Elastic License v2",
            "name": "test rule",
            "query": "process.name:test.query",
            "risk_score": 21,
            "rule_id": str(uuid.uuid4()),
            "severity": "low",
            "type": "query",
        }

        def build_rule(response_actions: list[dict[str, Any]]) -> TOMLRuleContents:
            """Helper function to build a TOMLRuleContents object."""
            metadata = {
                "creation_date": "1970/01/01",
                "updated_date": "1970/01/01",
                "min_stack_version": load_current_package_version(),
            }
            data = base_fields.copy()
            data["response_actions"] = response_actions
            return TOMLRuleContents.from_dict({"metadata": metadata, "rule": data})

        response_actions = [
            {
                "action_type_id": ".osquery",
                "params": {
                    "query": "select * from processes;",
                    "timeout": 120,
                    "ecs_mapping": {"process.pid": {"field": "pid"}},
                    "queries": [{"id": "processes", "query": "select * from processes;"}],
                },
            },
            {"action_type_id": ".endpoint", "params": {"command": "isolate", "comment": "isolate host"}},
            {
                "action_type_id": ".endpoint",
                "params": {
                    "command": "kill-process",
                    "comment": "kill process",
                    "config": {"field": "", "overwrite": True},
                },
            },
        ]
        contents = build_rule(response_actions)
        self.assertEqual(contents.to_api_format()["response_actions"], response_actions)


class TestVersionLockSchema(unittest.TestCase):
    """Test that the version lock has proper entries."""

    @classmethod
    def setUpClass(cls):
        cls.version_lock_contents = {
            "33f306e8-417c-411b-965c-c2812d6d3f4d": {
                "rule_name": "Remote File Download via PowerShell",
                "sha256": "8679cd72bf85b67dde3dcfdaba749ed1fa6560bca5efd03ed41c76a500ce31d6",
                "type": "eql",
                "version": 4,
            },
            "34fde489-94b0-4500-a76f-b8a157cf9269": {
                "min_stack_version": "8.2",
                "previous": {
                    "7.13": {
                        "rule_name": "Telnet Port Activity",
                        "sha256": "3dd4a438c915920e6ddb0a5212603af5d94fb8a6b51a32f223d930d7e3becb89",
                        "type": "query",
                        "version": 9,
                    }
                },
                "rule_name": "Telnet Port Activity",
                "sha256": "b0bdfa73639226fb83eadc0303ad1801e0707743f96a36209aa58228d3bf6a89",
                "type": "query",
                "version": 10,
            },
        }

    def test_version_lock_no_previous(self):
        """Pass field validation on version lock without nested previous fields"""
        version_lock_contents = copy.deepcopy(self.version_lock_contents)
        VersionLockFile.from_dict({"data": version_lock_contents})

    @unittest.skipIf(RULES_CONFIG.bypass_version_lock, "Version lock bypassed")
    def test_version_lock_has_nested_previous(self):
        """Fail field validation on version lock with nested previous fields"""
        version_lock_contents = copy.deepcopy(self.version_lock_contents)
        with self.assertRaises(ValidationError):
            previous = version_lock_contents["34fde489-94b0-4500-a76f-b8a157cf9269"]["previous"]
            version_lock_contents["34fde489-94b0-4500-a76f-b8a157cf9269"]["previous"]["previous"] = previous
            VersionLockFile.from_dict({"data": version_lock_contents})


class TestVersions(unittest.TestCase):
    """Test that schema versioning aligns."""

    def test_stack_schema_map(self):
        """Test to ensure that an entry exists in the stack-schema-map for the current package version."""
        package_version = Version.parse(load_current_package_version(), optional_minor_and_patch=True)
        stack_map = utils.load_etc_dump(["stack-schema-map.yaml"])
        err_msg = f"There is no entry defined for the current package ({package_version}) in the stack-schema-map"
        self.assertIn(package_version, [Version.parse(v) for v in stack_map], err_msg)


class TestESQLValidation(unittest.TestCase):
    """Test ESQL rule validation"""

    def test_esql_data_validation(self):
        """Test ESQL rule data validation"""

        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)

        # Most used order of the metadata fields
        query = """
            FROM logs-windows.powershell_operational* METADATA _id, _version, _index
            | WHERE event.code == "4104"
            | KEEP event.code, _id, _version, _index
        """
        rule_dict["rule"]["query"] = query
        _ = RuleCollection().load_dict(rule_dict, path=rule_path)

        # The order of the metadata fields from the example in the docs -
        # https://www.elastic.co/guide/en/security/8.17/rules-ui-create.html#esql-non-agg-query
        query = """
            FROM logs-windows.powershell_operational* METADATA _id, _index, _version
            | WHERE event.code == "4104"
            | KEEP event.code, _id, _version, _index
        """
        rule_dict["rule"]["query"] = query
        _ = RuleCollection().load_dict(rule_dict, path=rule_path)

        # Different metadata fields
        with pytest.raises(EsqlSemanticError):
            query = """
                FROM logs-windows.powershell_operational* METADATA _foo, _index
                | WHERE event.code == "4104"
                | KEEP event.code, _id, _version, _index
            """
            rule_dict["rule"]["query"] = query
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

        # Missing `keep`
        with pytest.raises(EsqlSemanticError):
            query = """
                FROM logs-windows.powershell_operational* METADATA _id, _index, _version
                | WHERE event.code == "4104"
            """
            rule_dict["rule"]["query"] = query
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

    def test_esql_keep_validation_bypass_missing_keep(self):
        """ES|QL keep checks are skipped when DR_BYPASS_ESQL_KEEP_VALIDATION is set."""
        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)
        query = """
            FROM logs-windows.powershell_operational* METADATA _id, _index, _version
            | WHERE event.code == "4104"
        """
        rule_dict["rule"]["query"] = query
        with unittest.mock.patch.dict(os.environ, {"DR_BYPASS_ESQL_KEEP_VALIDATION": "1"}):
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

    def test_esql_keep_bypass_does_not_skip_from_metadata_validation(self):
        """FROM METADATA requirement still applies when only keep validation is bypassed."""
        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)
        query = """
            FROM logs-windows.powershell_operational*
            | WHERE event.code == "4104"
        """
        rule_dict["rule"]["query"] = query
        with (
            unittest.mock.patch.dict(os.environ, {"DR_BYPASS_ESQL_KEEP_VALIDATION": "1"}),
            pytest.raises(EsqlSemanticError),
        ):
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

    def test_esql_metadata_validation_bypass_missing_from_metadata(self):
        """ES|QL FROM METADATA checks are skipped when DR_BYPASS_ESQL_METADATA_VALIDATION is set."""
        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)
        query = """
            FROM logs-windows.powershell_operational*
            | WHERE event.code == "4104"
            | KEEP event.code, _id, _version, _index
        """
        rule_dict["rule"]["query"] = query
        with unittest.mock.patch.dict(os.environ, {"DR_BYPASS_ESQL_METADATA_VALIDATION": "1"}):
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

    def test_esql_metadata_bypass_does_not_skip_keep_validation(self):
        """`keep` validation still applies when only FROM metadata validation is bypassed."""
        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)
        query = """
            FROM logs-windows.powershell_operational*
            | WHERE event.code == "4104"
        """
        rule_dict["rule"]["query"] = query
        with (
            unittest.mock.patch.dict(os.environ, {"DR_BYPASS_ESQL_METADATA_VALIDATION": "1"}),
            pytest.raises(EsqlSemanticError),
        ):
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)

    def test_esql_keep_validation_bypass_missing_metadata_in_keep(self):
        """ES|QL metadata-in-keep checks are skipped when DR_BYPASS_ESQL_KEEP_VALIDATION is set."""
        # A random ESQL rule to deliver a test query
        rule_path = Path("tests/data/command_control_dummy_production_rule.toml")
        rule_body = rule_path.read_text()
        rule_dict = RuleCollection.deserialize_toml_string(rule_body)
        query = """
            FROM logs-windows.powershell_operational* METADATA _id, _version, _index
            | WHERE event.code == "4104"
            | KEEP event.code
        """
        rule_dict["rule"]["query"] = query
        with unittest.mock.patch.dict(os.environ, {"DR_BYPASS_ESQL_KEEP_VALIDATION": "1"}):
            _ = RuleCollection().load_dict(rule_dict, path=rule_path)