diff --git a/detection_rules/beats.py b/detection_rules/beats.py
index 972c6da8134..feaa887acbb 100644
--- a/detection_rules/beats.py
+++ b/detection_rules/beats.py
@@ -117,6 +117,14 @@ def _flatten_schema(schema: list, prefix="") -> list:
             # it's probably not perfect, but we can fix other bugs as we run into them later
             if len(schema) == 1 and nested_prefix.startswith(prefix + prefix):
                 nested_prefix = s["name"] + "."
+            if "field" in s:
+                # integrations sometimes have a group with a single field
+                flattened.extend(_flatten_schema(s["field"], prefix=nested_prefix))
+                continue
+            elif "fields" not in s:
+                # integrations sometimes have a group with no fields
+                continue
+
             flattened.extend(_flatten_schema(s["fields"], prefix=nested_prefix))
         elif "fields" in s:
             flattened.extend(_flatten_schema(s["fields"], prefix=prefix))
@@ -131,6 +139,10 @@ def _flatten_schema(schema: list, prefix="") -> list:
     return flattened
 
 
+def flatten_ecs_schema(schema: dict) -> dict:
+    return _flatten_schema(schema)
+
+
 def get_field_schema(base_directory, prefix="", include_common=False):
     base_directory = base_directory.get("folders", {}).get("_meta", {}).get("files", {})
     flattened = []
diff --git a/detection_rules/devtools.py b/detection_rules/devtools.py
index a77b2533e74..41b98382e9b 100644
--- a/detection_rules/devtools.py
+++ b/detection_rules/devtools.py
@@ -33,7 +33,8 @@
 from .endgame import EndgameSchemaManager
 from .eswrap import CollectEvents, add_range_to_dsl
 from .ghwrap import GithubClient, update_gist
-from .integrations import build_integrations_manifest
+from .integrations import (build_integrations_manifest, build_integrations_schemas, find_latest_compatible_version,
+                           load_integrations_manifests)
 from .main import root
 from .misc import PYTHON_LICENSE, add_client, client_error
 from .packaging import (CURRENT_RELEASE_PATH, PACKAGE_FILE, RELEASE_DIR,
@@ -1174,10 +1175,48 @@ def integrations_group():
 def build_integration_manifests(overwrite: bool):
     """Builds consolidated integrations manifests file."""
     click.echo("loading rules to determine all integration tags")
+
+    def flatten(tag_list: List[str]) -> List[str]:
+        return list(set([tag for tags in tag_list for tag in (flatten(tags) if isinstance(tags, list) else [tags])]))
+
     rules = RuleCollection.default()
-    integration_tags = list(set([r.contents.metadata.integration for r in rules if r.contents.metadata.integration]))
-    click.echo(f"integration tags identified: {integration_tags}")
-    build_integrations_manifest(overwrite, integration_tags)
+    integration_tags = [r.contents.metadata.integration for r in rules if r.contents.metadata.integration]
+    unique_integration_tags = flatten(integration_tags)
+    click.echo(f"integration tags identified: {unique_integration_tags}")
+    build_integrations_manifest(overwrite, unique_integration_tags)
+
+
+@integrations_group.command('build-schemas')
+@click.option('--overwrite', '-o', is_flag=True, help="Overwrite the entire integrations-schema.json.gz file")
+def build_integration_schemas(overwrite: bool):
+    """Builds consolidated integrations schemas file."""
+    click.echo("Building integration schemas...")
+
+    start_time = time.perf_counter()
+    build_integrations_schemas(overwrite)
+    end_time = time.perf_counter()
+    click.echo(f"Time taken to generate schemas: {(end_time - start_time)/60:.2f} minutes")
+
+
+@integrations_group.command('show-latest-compatible')
+@click.option('--package', '-p', help='Name of package')
+@click.option('--stack_version', '-s', required=True, help='Rule stack version')
+def show_latest_compatible_version(package: str, stack_version: str) -> None:
+    """Prints the latest integration compatible version for specified package based on stack version supplied."""
+
+    packages_manifest = None
+    try:
+        packages_manifest = load_integrations_manifests()
+    except Exception as e:
+        click.echo(f"Error loading integrations manifests: {str(e)}")
+        return
+
+    try:
+        version = find_latest_compatible_version(package, "", stack_version, packages_manifest)
+        click.echo(f"Compatible integration {version=}")
+    except Exception as e:
+        click.echo(f"Error finding compatible version: {str(e)}")
+        return
 
 
 @dev_group.group('schemas')
diff --git a/detection_rules/etc/integration-manifests.json.gz b/detection_rules/etc/integration-manifests.json.gz
index aab5ce6c9f1..a741b6a5606 100644
Binary files a/detection_rules/etc/integration-manifests.json.gz and b/detection_rules/etc/integration-manifests.json.gz differ
diff --git a/detection_rules/etc/integration-schemas.json.gz b/detection_rules/etc/integration-schemas.json.gz
new file mode 100644
index 00000000000..57d56fb4f43
Binary files /dev/null and b/detection_rules/etc/integration-schemas.json.gz differ
diff --git a/detection_rules/integrations.py b/detection_rules/integrations.py
index 94b837ad137..6edda773f64 100644
--- a/detection_rules/integrations.py
+++ b/detection_rules/integrations.py
@@ -4,20 +4,27 @@
 # 2.0.
 
 """Functions to support and interact with Kibana integrations."""
+import glob
 import gzip
 import json
-import os
 import re
 from collections import OrderedDict
 from pathlib import Path
+from typing import Generator, Tuple, Union
 
 import requests
+import yaml
 from marshmallow import EXCLUDE, Schema, fields, post_load
 
+import kql
+
+from . import ecs
+from .beats import flatten_ecs_schema
 from .semver import Version
-from .utils import cached, get_etc_path, read_gzip
+from .utils import cached, get_etc_path, read_gzip, unzip
 
 MANIFEST_FILE_PATH = Path(get_etc_path('integration-manifests.json.gz'))
+SCHEMA_FILE_PATH = Path(get_etc_path('integration-schemas.json.gz'))
 
 
 @cached
@@ -26,11 +33,18 @@ def load_integrations_manifests() -> dict:
     return json.loads(read_gzip(get_etc_path('integration-manifests.json.gz')))
 
 
+@cached
+def load_integrations_schemas() -> dict:
+    """Load the consolidated integrations schemas."""
+    return json.loads(read_gzip(get_etc_path('integration-schemas.json.gz')))
+
+
 class IntegrationManifestSchema(Schema):
     name = fields.Str(required=True)
     version = fields.Str(required=True)
     release = fields.Str(required=True)
     description = fields.Str(required=True)
+    download = fields.Str(required=True)
     conditions = fields.Dict(required=True)
     policy_templates = fields.List(fields.Dict, required=True)
     owner = fields.Dict(required=False)
@@ -44,8 +58,8 @@ def transform_policy_template(self, data, **kwargs):
 def build_integrations_manifest(overwrite: bool, rule_integrations: list) -> None:
     """Builds a new local copy of manifest.yaml from integrations Github."""
     if overwrite:
-        if os.path.exists(MANIFEST_FILE_PATH):
-            os.remove(MANIFEST_FILE_PATH)
+        if MANIFEST_FILE_PATH.exists():
+            MANIFEST_FILE_PATH.unlink()
 
     final_integration_manifests = {integration: {} for integration in rule_integrations}
 
@@ -62,6 +76,63 @@ def build_integrations_manifest(overwrite: bool, rule_integrations: list) -> Non
     print(f"final integrations manifests dumped: {MANIFEST_FILE_PATH}")
 
 
+def build_integrations_schemas(overwrite: bool) -> None:
+    """Builds a new local copy of integration-schemas.json.gz from EPR integrations."""
+
+    final_integration_schemas = {}
+    saved_integration_schemas = {}
+
+    # Check if the file already exists and handle accordingly
+    if overwrite and SCHEMA_FILE_PATH.exists():
+        SCHEMA_FILE_PATH.unlink()
+    elif SCHEMA_FILE_PATH.exists():
+        saved_integration_schemas = load_integrations_schemas()
+
+    # Load the integration manifests
+    integration_manifests = load_integrations_manifests()
+
+    # Loop through the packages and versions
+    for package, versions in integration_manifests.items():
+        print(f"processing {package}")
+        final_integration_schemas.setdefault(package, {})
+        for version, manifest in versions.items():
+            if package in saved_integration_schemas and version in saved_integration_schemas[package]:
+                continue
+
+            # Download the zip file
+            download_url = f"https://epr.elastic.co{manifest['download']}"
+            response = requests.get(download_url)
+            response.raise_for_status()
+
+            # Update the final integration schemas
+            final_integration_schemas[package].update({version: {}})
+
+            # Open the zip file
+            with unzip(response.content) as zip_ref:
+                for file in zip_ref.namelist():
+                    # Check if the file is a match
+                    if glob.fnmatch.fnmatch(file, '*/fields/*.yml'):
+                        integration_name = Path(file).parent.parent.name
+                        final_integration_schemas[package][version].setdefault(integration_name, {})
+                        file_data = zip_ref.read(file)
+                        schema_fields = yaml.safe_load(file_data)
+
+                        # Parse the schema and add to the integration_manifests
+                        data = flatten_ecs_schema(schema_fields)
+                        flat_data = {field['name']: field['type'] for field in data}
+
+                        final_integration_schemas[package][version][integration_name].update(flat_data)
+
+                        del file_data
+
+    # Write the final integration schemas to disk
+    with gzip.open(SCHEMA_FILE_PATH, "w") as schema_file:
+        schema_file_bytes = json.dumps(final_integration_schemas).encode("utf-8")
+        schema_file.write(schema_file_bytes)
+
+    print(f"final integrations manifests dumped: {SCHEMA_FILE_PATH}")
+
+
 def find_least_compatible_version(package: str, integration: str,
                                   current_stack_version: str, packages_manifest: dict) -> str:
     """Finds least compatible version for specified integration based on stack version supplied."""
@@ -89,12 +160,54 @@ def find_least_compatible_version(package: str, integration: str,
     raise ValueError(f"no compatible version for integration {package}:{integration}")
 
 
+def find_latest_compatible_version(package: str, integration: str,
+                                   rule_stack_version: str, packages_manifest: dict) -> Union[None, Tuple[str, str]]:
+    """Finds least compatible version for specified integration based on stack version supplied."""
+
+    if not package:
+        raise ValueError("Package must be specified")
+
+    package_manifest = packages_manifest.get(package)
+    if package_manifest is None:
+        raise ValueError(f"Package {package} not found in manifest.")
+
+    # Converts the dict keys (version numbers) to Version objects for proper sorting (descending)
+    integration_manifests = sorted(package_manifest.items(), key=lambda x: Version(str(x[0])), reverse=True)
+    notice = ""
+
+    for version, manifest in integration_manifests:
+        kibana_conditions = manifest.get("conditions", {}).get("kibana", {})
+        version_requirement = kibana_conditions.get("version")
+        if not version_requirement:
+            raise ValueError(f"Manifest for {package}:{integration} version {version} is missing conditions.")
+
+        compatible_versions = re.sub(r"\>|\<|\=|\^", "", version_requirement).split(" || ")
+
+        if not compatible_versions:
+            raise ValueError(f"Manifest for {package}:{integration} version {version} is missing compatible versions")
+
+        highest_compatible_version = max(compatible_versions, key=lambda x: Version(x))
+
+        if Version(highest_compatible_version) > Version(rule_stack_version):
+            # generate notice message that a later integration version is available
+            integration = f" {integration.strip()}" if integration else ""
+
+            notice = (f"There is a new integration {package}{integration} version {version} available!",
+                      f"Update the rule min_stack version from {rule_stack_version} to "
+                      f"{highest_compatible_version} if using new features in this latest version.")
+
+        elif int(highest_compatible_version[0]) == int(rule_stack_version[0]):
+            return version, notice
+
+    raise ValueError(f"no compatible version for integration {package}:{integration}")
+
+
 def get_integration_manifests(integration: str) -> list:
     """Iterates over specified integrations from package-storage and combines manifests per version."""
     epr_search_url = "https://epr.elastic.co/search"
 
     # link for search parameters - https://github.com/elastic/package-registry
-    epr_search_parameters = {"package": f"{integration}", "prerelease": "true",
+    epr_search_parameters = {"package": f"{integration}", "prerelease": "false",
                              "all": "true", "include_policy_templates": "true"}
     epr_search_response = requests.get(epr_search_url, params=epr_search_parameters)
     epr_search_response.raise_for_status()
@@ -106,3 +219,63 @@ def get_integration_manifests(integration: str) -> list:
     print(f"loaded {integration} manifests from the following package versions: "
           f"{[manifest['version'] for manifest in manifests]}")
     return manifests
+
+
+def get_integration_schema_data(data, meta, package_integrations: dict) -> Generator[dict, None, None]:
+    """Iterates over specified integrations from package-storage and combines schemas per version."""
+
+    # lazy import to avoid circular import
+    from .rule import (  # pylint: disable=import-outside-toplevel
+        QueryRuleData, RuleMeta
+    )
+
+    data: QueryRuleData = data
+    meta: RuleMeta = meta
+
+    packages_manifest = load_integrations_manifests()
+    integrations_schemas = load_integrations_schemas()
+
+    # validate the query against related integration fields
+    if isinstance(data, QueryRuleData) and data.language != 'lucene' and meta.maturity == "production":
+
+        # flag to only warn once per integration for available upgrades
+        notify_update_available = True
+
+        for stack_version, mapping in meta.get_validation_stack_versions().items():
+            ecs_version = mapping['ecs']
+            endgame_version = mapping['endgame']
+
+            ecs_schema = ecs.flatten_multi_fields(ecs.get_schema(ecs_version, name='ecs_flat'))
+
+            for pk_int in package_integrations:
+                package = pk_int["package"]
+                integration = pk_int["integration"]
+
+                package_version, notice = find_latest_compatible_version(package=package,
+                                                                         integration=integration,
+                                                                         rule_stack_version=meta.min_stack_version,
+                                                                         packages_manifest=packages_manifest)
+
+                if notify_update_available and notice and data.get("notify", False):
+                    # Notify for now, as to not lock rule stacks to integrations
+                    notify_update_available = False
+                    print(f"\n{data.get('name')}")
+                    print(*notice)
+
+                schema = {}
+                if integration is None:
+                    # Use all fields from each dataset
+                    for dataset in integrations_schemas[package][package_version]:
+                        schema.update(integrations_schemas[package][package_version][dataset])
+                else:
+                    if integration not in integrations_schemas[package][package_version]:
+                        raise ValueError(f"Integration {integration} not found in package {package} "
+                                         f"version {package_version}")
+                    schema = integrations_schemas[package][package_version][integration]
+                schema.update(ecs_schema)
+                integration_schema = {k: kql.parser.elasticsearch_type_family(v) for k, v in schema.items()}
+
+                data = {"schema": integration_schema, "package": package, "integration": integration,
+                        "stack_version": stack_version, "ecs_version": ecs_version,
+                        "package_version": package_version, "endgame_version": endgame_version}
+                yield data
diff --git a/detection_rules/rule.py b/detection_rules/rule.py
index 4946d928b54..1121464ee86 100644
--- a/detection_rules/rule.py
+++ b/detection_rules/rule.py
@@ -240,6 +240,10 @@ def get_restricted_fields(self) -> Optional[Dict[str, tuple]]:
     def data_validator(self) -> Optional['DataValidator']:
         return DataValidator(is_elastic_rule=self.is_elastic_rule, **self.to_dict())
 
+    @cached_property
+    def notify(self) -> bool:
+        return os.environ.get('DR_NOTIFY_INTEGRATION_UPDATE_AVAILABLE') is not None
+
     @cached_property
     def parsed_note(self) -> Optional[MarkoDocument]:
         dv = self.data_validator
@@ -847,7 +851,7 @@ def _add_related_integrations(self, obj: dict) -> None:
 
             if self.check_restricted_field_version(field_name):
                 if isinstance(self.data, QueryRuleData) and self.data.language != 'lucene':
-                    package_integrations = self._get_packaged_integrations(packages_manifest)
+                    package_integrations = self.get_packaged_integrations(self.data, self.metadata, packages_manifest)
 
                     if not package_integrations:
                         return
@@ -947,11 +951,13 @@ def compare_field_versions(min_stack: Version, max_stack: Version) -> bool:
         max_stack = max_stack or current_version
         return Version(min_stack) <= current_version >= Version(max_stack)
 
-    def _get_packaged_integrations(self, package_manifest: dict) -> Optional[List[dict]]:
+    @classmethod
+    def get_packaged_integrations(cls, data: QueryRuleData, meta: RuleMeta,
+                                  package_manifest: dict) -> Optional[List[dict]]:
         packaged_integrations = []
         datasets = set()
 
-        for node in self.data.get('ast', []):
+        for node in data.get('ast', []):
             if isinstance(node, eql.ast.Comparison) and str(node.left) == 'event.dataset':
                 datasets.update(set(n.value for n in node if isinstance(n, eql.ast.Literal)))
             elif isinstance(node, FieldComparison) and str(node.field) == 'event.dataset':
@@ -960,10 +966,10 @@ def _get_packaged_integrations(self, package_manifest: dict) -> Optional[List[di
         if not datasets:
             # windows and endpoint integration do not have event.dataset fields in queries
             # integration is None to remove duplicate references upstream in Kibana
-            rule_integrations = self.metadata.get("integration", [])
+            rule_integrations = meta.get("integration", [])
             if rule_integrations:
                 for integration in rule_integrations:
-                    if integration in ["windows", "endpoint", "apm"]:
+                    if integration in definitions.NON_DATASET_PACKAGES:
                         packaged_integrations.append({"package": integration, "integration": None})
 
         for value in sorted(datasets):
diff --git a/detection_rules/rule_validators.py b/detection_rules/rule_validators.py
index 332955eb116..e2ae9f2177c 100644
--- a/detection_rules/rule_validators.py
+++ b/detection_rules/rule_validators.py
@@ -8,14 +8,16 @@
 from typing import List, Optional, Union
 
 import eql
+
 import kql
 
 from . import ecs, endgame
-from .rule import QueryRuleData, QueryValidator, RuleMeta
+from .integrations import get_integration_schema_data, load_integrations_manifests
+from .rule import QueryRuleData, QueryValidator, RuleMeta, TOMLRuleContents
 
 
 class KQLValidator(QueryValidator):
-    """Specific fields for query event types."""
+    """Specific fields for KQL query event types."""
 
     @cached_property
     def ast(self) -> kql.ast.Expression:
@@ -34,29 +36,97 @@ def validate(self, data: QueryRuleData, meta: RuleMeta) -> None:
             # syntax only, which is done via self.ast
             return
 
-        for stack_version, mapping in meta.get_validation_stack_versions().items():
-            beats_version = mapping['beats']
-            ecs_version = mapping['ecs']
-            err_trailer = f'stack: {stack_version}, beats: {beats_version}, ecs: {ecs_version}'
+        if isinstance(data, QueryRuleData) and data.language != 'lucene':
+            packages_manifest = load_integrations_manifests()
+            package_integrations = TOMLRuleContents.get_packaged_integrations(data, meta, packages_manifest)
+
+            if package_integrations:
+                # validate the query against related integration fields
+                self.validate_integration(data, meta, package_integrations)
+            else:
+                for stack_version, mapping in meta.get_validation_stack_versions().items():
+                    beats_version = mapping['beats']
+                    ecs_version = mapping['ecs']
+                    err_trailer = f'stack: {stack_version}, beats: {beats_version}, ecs: {ecs_version}'
+
+                    beat_types, beat_schema, schema = self.get_beats_schema(data.index or [],
+                                                                            beats_version, ecs_version)
+
+                    try:
+                        kql.parse(self.query, schema=schema)
+                    except kql.KqlParseError as exc:
+                        message = exc.error_msg
+                        trailer = err_trailer
+                        if "Unknown field" in message and beat_types:
+                            trailer = f"\nTry adding event.module or event.dataset to specify beats module\n\n{trailer}"
 
-            beat_types, beat_schema, schema = self.get_beats_schema(data.index or [], beats_version, ecs_version)
+                        raise kql.KqlParseError(exc.error_msg, exc.line, exc.column, exc.source,
+                                                len(exc.caret.lstrip()), trailer=trailer) from None
+                    except Exception:
+                        print(err_trailer)
+                        raise
+
+    def validate_integration(self, data: QueryRuleData, meta: RuleMeta, package_integrations: List[dict]) -> None:
+        """Validate the query, called from the parent which contains [metadata] information."""
+        if meta.query_schema_validation is False or meta.maturity == "deprecated":
+            # syntax only, which is done via self.ast
+            return
+
+        error_fields = {}
+        current_stack_version = ""
+        combined_schema = {}
+        for integration_schema_data in get_integration_schema_data(data, meta, package_integrations):
+            ecs_version = integration_schema_data['ecs_version']
+            integration = integration_schema_data['integration']
+            package = integration_schema_data['package']
+            package_version = integration_schema_data['package_version']
+            integration_schema = integration_schema_data['schema']
+            stack_version = integration_schema_data['stack_version']
+
+            if stack_version != current_stack_version:
+                # reset the combined schema for each stack version
+                current_stack_version = stack_version
+                combined_schema = {}
+
+            # add non-ecs-schema fields for edge cases not added to the integration
+            for index_name in data.index:
+                integration_schema.update(**ecs.flatten(ecs.get_index_schema(index_name)))
+            combined_schema.update(**integration_schema)
 
             try:
-                kql.parse(self.query, schema=schema)
+                # validate the query against the integration fields with the package version
+                kql.parse(self.query, schema=integration_schema)
             except kql.KqlParseError as exc:
-                message = exc.error_msg
-                trailer = err_trailer
-                if "Unknown field" in message and beat_types:
-                    trailer = f"\nTry adding event.module or event.dataset to specify beats module\n\n{trailer}"
+                if exc.error_msg == "Unknown field":
+                    field = extract_error_field(exc)
+                    trailer = (f"\n\tTry adding event.module or event.dataset to specify integration module\n\t"
+                               f"Will check against integrations {meta.integration} combined.\n\t"
+                               f"{package=}, {integration=}, {package_version=}, "
+                               f"{stack_version=}, {ecs_version=}"
+                               )
+                    error_fields[field] = {"error": exc, "trailer": trailer}
+                    print(f"\nWarning: `{field}` in `{data.name}` not found in schema. {trailer}")
+                else:
+                    raise kql.KqlParseError(exc.error_msg, exc.line, exc.column, exc.source,
+                                            len(exc.caret.lstrip()), trailer=trailer) from None
+
+        # don't error on fields that are in another integration schema
+        for field in list(error_fields.keys()):
+            if field in combined_schema:
+                del error_fields[field]
+
+        # raise the first error
+        if error_fields:
+            _, data = next(iter(error_fields.items()))
+            exc = data["error"]
+            trailer = data["trailer"]
 
-                raise kql.KqlParseError(exc.error_msg, exc.line, exc.column, exc.source,
-                                        len(exc.caret.lstrip()), trailer=trailer) from None
-            except Exception:
-                print(err_trailer)
-                raise
+            raise kql.KqlParseError(exc.error_msg, exc.line, exc.column, exc.source,
+                                    len(exc.caret.lstrip()), trailer=trailer) from None
 
 
 class EQLValidator(QueryValidator):
+    """Specific fields for EQL query event types."""
 
     @cached_property
     def ast(self) -> eql.ast.Expression:
@@ -74,7 +144,108 @@ def text_fields(self, eql_schema: Union[ecs.KqlSchema2Eql, endgame.EndgameSchema
     def unique_fields(self) -> List[str]:
         return list(set(str(f) for f in self.ast if isinstance(f, eql.ast.Field)))
 
-    def validate_query_with_schema(self, schema: Union[ecs.KqlSchema2Eql, endgame.EndgameSchema],
+    def validate(self, data: 'QueryRuleData', meta: RuleMeta) -> None:
+        """Validate an EQL query while checking TOMLRule."""
+        if meta.query_schema_validation is False or meta.maturity == "deprecated":
+            # syntax only, which is done via self.ast
+            return
+
+        if isinstance(data, QueryRuleData) and data.language != 'lucene':
+            packages_manifest = load_integrations_manifests()
+            package_integrations = TOMLRuleContents.get_packaged_integrations(data, meta, packages_manifest)
+
+            if package_integrations:
+                # validate the query against related integration fields
+                self.validate_integration(data, meta, package_integrations)
+
+            else:
+                for stack_version, mapping in meta.get_validation_stack_versions().items():
+                    beats_version = mapping['beats']
+                    ecs_version = mapping['ecs']
+                    endgame_version = mapping['endgame']
+                    err_trailer = f'stack: {stack_version}, beats: {beats_version},' \
+                                  f'ecs: {ecs_version}, endgame: {endgame_version}'
+
+                    beat_types, beat_schema, schema = self.get_beats_schema(data.index or [],
+                                                                            beats_version, ecs_version)
+                    endgame_schema = self.get_endgame_schema(data.index, endgame_version)
+                    eql_schema = ecs.KqlSchema2Eql(schema)
+
+                    # validate query against the beats and eql schema
+                    self.validate_query_with_schema(data=data, schema=eql_schema, err_trailer=err_trailer,
+                                                    beat_types=beat_types)
+
+                    if endgame_schema:
+                        # validate query against the endgame schema
+                        self.validate_query_with_schema(data=data, schema=endgame_schema, err_trailer=err_trailer)
+
+    def validate_integration(self, data: QueryRuleData, meta: RuleMeta, package_integrations: List[dict]) -> None:
+        """Validate an EQL query while checking TOMLRule against integration schemas."""
+        if meta.query_schema_validation is False or meta.maturity == "deprecated":
+            # syntax only, which is done via self.ast
+            return
+
+        error_fields = {}
+        current_stack_version = ""
+        combined_schema = {}
+        for integration_schema_data in get_integration_schema_data(data, meta, package_integrations):
+            ecs_version = integration_schema_data['ecs_version']
+            integration = integration_schema_data['integration']
+            package = integration_schema_data['package']
+            package_version = integration_schema_data['package_version']
+            integration_schema = integration_schema_data['schema']
+            stack_version = integration_schema_data['stack_version']
+            endgame_version = integration_schema_data['endgame_version']
+
+            if stack_version != current_stack_version:
+                # reset the combined schema for each stack version
+                current_stack_version = stack_version
+                combined_schema = {}
+
+            # add non-ecs-schema fields for edge cases not added to the integration
+            for index_name in data.index:
+                integration_schema.update(**ecs.flatten(ecs.get_index_schema(index_name)))
+            combined_schema.update(**integration_schema)
+
+            eql_schema = ecs.KqlSchema2Eql(integration_schema)
+            err_trailer = f'stack: {stack_version}, integration: {integration},' \
+                          f'ecs: {ecs_version}, package: {package}, package_version: {package_version}'
+
+            try:
+                self.validate_query_with_schema(data=data, schema=eql_schema, err_trailer=err_trailer)
+            except eql.EqlParseError as exc:
+                message = exc.error_msg
+                if message == "Unknown field" or "Field not recognized" in message:
+                    field = extract_error_field(exc)
+                    trailer = (f"\n\tTry adding event.module or event.dataset to specify integration module\n\t"
+                               f"Will check against integrations {meta.integration} combined.\n\t"
+                               f"{package=}, {integration=}, {package_version=}, "
+                               f"{stack_version=}, {ecs_version=}"
+                               )
+                    error_fields[field] = {"error": exc, "trailer": trailer}
+                    print(f"\nWarning: `{field}` in `{data.name}` not found in schema. {trailer}")
+                else:
+                    raise exc
+
+            # Still need to check endgame if it's in the index
+            endgame_schema = self.get_endgame_schema(data.index, endgame_version)
+            if endgame_schema:
+                # validate query against the endgame schema
+                err_trailer = f'stack: {stack_version}, endgame: {endgame_version}'
+                self.validate_query_with_schema(data=data, schema=endgame_schema, err_trailer=err_trailer)
+
+        # don't error on fields that are in another integration schema
+        for field in list(error_fields.keys()):
+            if field in combined_schema:
+                del error_fields[field]
+
+        # raise the first error
+        if error_fields:
+            _, data = next(iter(error_fields.items()))
+            exc = data["error"]
+            raise exc
+
+    def validate_query_with_schema(self, data: 'QueryRuleData', schema: Union[ecs.KqlSchema2Eql, endgame.EndgameSchema],
                                    err_trailer: str, beat_types: list = None) -> None:
         """Validate the query against the schema."""
         try:
@@ -93,37 +264,17 @@ def validate_query_with_schema(self, schema: Union[ecs.KqlSchema2Eql, endgame.En
 
             raise exc.__class__(exc.error_msg, exc.line, exc.column, exc.source,
                                 len(exc.caret.lstrip()), trailer=trailer) from None
+
         except Exception:
             print(err_trailer)
             raise
 
-    def validate(self, data: 'QueryRuleData', meta: RuleMeta) -> None:
-        """Validate an EQL query while checking TOMLRule."""
-        if meta.query_schema_validation is False or meta.maturity == "deprecated":
-            # syntax only, which is done via self.ast
-            return
-
-        for stack_version, mapping in meta.get_validation_stack_versions().items():
-            beats_version = mapping['beats']
-            ecs_version = mapping['ecs']
-            endgame_version = mapping['endgame']
-            err_trailer = f'stack: {stack_version}, beats: {beats_version},' \
-                          f'ecs: {ecs_version}, endgame: {endgame_version}'
-
-            beat_types, beat_schema, schema = self.get_beats_schema(data.index or [], beats_version, ecs_version)
-            endgame_schema = self.get_endgame_schema(data.index, endgame_version)
-            eql_schema = ecs.KqlSchema2Eql(schema)
-
-            # validate query against the beats and eql schema
-            self.validate_query_with_schema(schema=eql_schema, err_trailer=err_trailer, beat_types=beat_types)
-
-            if endgame_schema:
-                # validate query against the endgame schema
-                self.validate_query_with_schema(schema=endgame_schema, err_trailer=err_trailer)
-
 
 def extract_error_field(exc: Union[eql.EqlParseError, kql.KqlParseError]) -> Optional[str]:
-    line = exc.source.splitlines()[exc.line]
+    """Extract the field name from an EQL or KQL parse error."""
+    lines = exc.source.splitlines()
+    mod = -1 if exc.line == len(lines) else 0
+    line = lines[exc.line + mod]
     start = exc.column
     stop = start + len(exc.caret.strip())
     return line[start:stop]
diff --git a/detection_rules/schemas/definitions.py b/detection_rules/schemas/definitions.py
index fb4bb0c8646..a012e39f0a6 100644
--- a/detection_rules/schemas/definitions.py
+++ b/detection_rules/schemas/definitions.py
@@ -27,6 +27,7 @@
 MINOR_SEMVER = r'^\d+\.\d+$'
 BRANCH_PATTERN = f'{VERSION_PATTERN}|^master$'
 
+NON_DATASET_PACKAGES = ['apm', 'endpoint', 'system', 'windows']
 INTERVAL_PATTERN = r'^\d+[mshd]$'
 TACTIC_URL = r'^https://attack.mitre.org/tactics/TA[0-9]+/$'
 TECHNIQUE_URL = r'^https://attack.mitre.org/techniques/T[0-9]+/$'
diff --git a/detection_rules/utils.py b/detection_rules/utils.py
index 234d64cd0d3..a09290abbfa 100644
--- a/detection_rules/utils.py
+++ b/detection_rules/utils.py
@@ -113,7 +113,7 @@ def load_etc_dump(*path):
 
 
 def save_etc_dump(contents, *path, **kwargs):
-    """Load a json/yml/toml file from the detection_rules/etc/ folder."""
+    """Save a json/yml/toml file from the detection_rules/etc/ folder."""
     path = get_etc_path(*path)
     _, ext = os.path.splitext(path)
     sort_keys = kwargs.pop('sort_keys', True)
diff --git a/rules/integrations/azure/initial_access_consent_grant_attack_via_azure_registered_application.toml b/rules/integrations/azure/initial_access_consent_grant_attack_via_azure_registered_application.toml
index 2b405d7db42..15ae654d376 100644
--- a/rules/integrations/azure/initial_access_consent_grant_attack_via_azure_registered_application.toml
+++ b/rules/integrations/azure/initial_access_consent_grant_attack_via_azure_registered_application.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/09/01"
-integration = ["azure"]
+integration = ["azure", "o365"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/14"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
@@ -14,7 +14,7 @@ permissions to an application. An adversary may create an Azure-registered appli
 as contact information, email, or documents.
 """
 from = "now-25m"
-index = ["filebeat-*", "logs-azure*"]
+index = ["filebeat-*", "logs-azure*", "logs-o365*"]
 language = "kuery"
 license = "Elastic License v2"
 name = "Possible Consent Grant Attack via Azure-Registered Application"
@@ -79,6 +79,7 @@ tags = [
     "Cloud",
     "Azure",
     "Continuous Monitoring",
+    "Microsoft 365",
     "SecOps",
     "Identity and Access",
     "Investigation Guide",
diff --git a/rules/linux/credential_access_bruteforce_password_guessing.toml b/rules/linux/credential_access_bruteforce_password_guessing.toml
index 8d5811c9914..650ff7723c1 100644
--- a/rules/linux/credential_access_bruteforce_password_guessing.toml
+++ b/rules/linux/credential_access_bruteforce_password_guessing.toml
@@ -1,9 +1,10 @@
 [metadata]
 creation_date = "2022/09/14"
+integration = ["system"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/11/28"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/linux/credential_access_potential_linux_ssh_bruteforce.toml b/rules/linux/credential_access_potential_linux_ssh_bruteforce.toml
index c2b48ce10a6..cf8ad0e1fb2 100644
--- a/rules/linux/credential_access_potential_linux_ssh_bruteforce.toml
+++ b/rules/linux/credential_access_potential_linux_ssh_bruteforce.toml
@@ -1,9 +1,10 @@
 [metadata]
 creation_date = "2022/09/14"
+integration = ["system"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/11/28"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/linux/credential_access_potential_linux_ssh_bruteforce_root.toml b/rules/linux/credential_access_potential_linux_ssh_bruteforce_root.toml
index b1a6fcb87f7..bf9d7931f76 100644
--- a/rules/linux/credential_access_potential_linux_ssh_bruteforce_root.toml
+++ b/rules/linux/credential_access_potential_linux_ssh_bruteforce_root.toml
@@ -1,9 +1,10 @@
 [metadata]
 creation_date = "2022/09/14"
+integration = ["system"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2023/01/27"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/collection_email_outlook_mailbox_via_com.toml b/rules/windows/collection_email_outlook_mailbox_via_com.toml
index 16a21357df6..8cdefbd39bf 100644
--- a/rules/windows/collection_email_outlook_mailbox_via_com.toml
+++ b/rules/windows/collection_email_outlook_mailbox_via_com.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2023/01/11"
-integration = ["windows"]
+integration = ["endpoint"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.4.0"
-updated_date = "2023/01/11"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
@@ -28,8 +28,8 @@ timestamp_override = "event.ingested"
 type = "eql"
 
 query = '''
-process where event.action == "start" and process.name : "OUTLOOK.EXE"  and 
- process.Ext.effective_parent.name != null and 
+process where event.action == "start" and process.name : "OUTLOOK.EXE"  and
+ process.Ext.effective_parent.name != null and
  not process.Ext.effective_parent.executable : ("?:\\Program Files\\*", "?:\\Program Files (x86)\\*")
 '''
 
@@ -56,14 +56,14 @@ reference = "https://attack.mitre.org/tactics/TA0009/"
 [[rule.threat]]
 framework = "MITRE ATT&CK"
 
-[[rule.threat.technique]] 
-id = "T1559" 
-name = "Inter-Process Communication" 
+[[rule.threat.technique]]
+id = "T1559"
+name = "Inter-Process Communication"
 reference = "https://attack.mitre.org/techniques/T1559/"
 
-[[rule.threat.technique.subtechnique]] 
+[[rule.threat.technique.subtechnique]]
 id = "T1559.001"
-name = "Component Object Model" 
+name = "Component Object Model"
 reference = "https://attack.mitre.org/techniques/T1559/001/"
 
 
diff --git a/rules/windows/credential_access_bruteforce_admin_account.toml b/rules/windows/credential_access_bruteforce_admin_account.toml
index dd2477415c5..16ad3fe9ed6 100644
--- a/rules/windows/credential_access_bruteforce_admin_account.toml
+++ b/rules/windows/credential_access_bruteforce_admin_account.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_bruteforce_multiple_logon_failure_followed_by_success.toml b/rules/windows/credential_access_bruteforce_multiple_logon_failure_followed_by_success.toml
index e8c75e76b3e..121fbbacf2e 100644
--- a/rules/windows/credential_access_bruteforce_multiple_logon_failure_followed_by_success.toml
+++ b/rules/windows/credential_access_bruteforce_multiple_logon_failure_followed_by_success.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_bruteforce_multiple_logon_failure_same_srcip.toml b/rules/windows/credential_access_bruteforce_multiple_logon_failure_same_srcip.toml
index d0a7f843322..50a7108c4fa 100644
--- a/rules/windows/credential_access_bruteforce_multiple_logon_failure_same_srcip.toml
+++ b/rules/windows/credential_access_bruteforce_multiple_logon_failure_same_srcip.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_dcsync_replication_rights.toml b/rules/windows/credential_access_dcsync_replication_rights.toml
index c162a1bd43d..a029e876fbe 100644
--- a/rules/windows/credential_access_dcsync_replication_rights.toml
+++ b/rules/windows/credential_access_dcsync_replication_rights.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/08"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2023/01/27"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_disable_kerberos_preauth.toml b/rules/windows/credential_access_disable_kerberos_preauth.toml
index ecd5112ff74..3735f99e122 100644
--- a/rules/windows/credential_access_disable_kerberos_preauth.toml
+++ b/rules/windows/credential_access_disable_kerberos_preauth.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/01/24"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_ldap_attributes.toml b/rules/windows/credential_access_ldap_attributes.toml
index 48003aecea8..4a28b83ce6b 100644
--- a/rules/windows/credential_access_ldap_attributes.toml
+++ b/rules/windows/credential_access_ldap_attributes.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/11/09"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_lsass_memdump_handle_access.toml b/rules/windows/credential_access_lsass_memdump_handle_access.toml
index 96065c6ec3d..cd7356f4146 100644
--- a/rules/windows/credential_access_lsass_memdump_handle_access.toml
+++ b/rules/windows/credential_access_lsass_memdump_handle_access.toml
@@ -1,8 +1,8 @@
 [metadata]
 creation_date = "2022/02/16"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
-updated_date = "2023/01/03"
+updated_date = "2023/02/01"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
 
diff --git a/rules/windows/credential_access_remote_sam_secretsdump.toml b/rules/windows/credential_access_remote_sam_secretsdump.toml
index fef0a1e305c..84e355e24a8 100644
--- a/rules/windows/credential_access_remote_sam_secretsdump.toml
+++ b/rules/windows/credential_access_remote_sam_secretsdump.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/03/01"
-integration = ["endpoint", "windows"]
+integration = ["endpoint", "system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2023/01/31"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_saved_creds_vault_winlog.toml b/rules/windows/credential_access_saved_creds_vault_winlog.toml
index 4820861148f..840624ca657 100644
--- a/rules/windows/credential_access_saved_creds_vault_winlog.toml
+++ b/rules/windows/credential_access_saved_creds_vault_winlog.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/30"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_seenabledelegationprivilege_assigned_to_user.toml b/rules/windows/credential_access_seenabledelegationprivilege_assigned_to_user.toml
index 5e932829435..74b06c7016f 100644
--- a/rules/windows/credential_access_seenabledelegationprivilege_assigned_to_user.toml
+++ b/rules/windows/credential_access_seenabledelegationprivilege_assigned_to_user.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/01/27"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_shadow_credentials.toml b/rules/windows/credential_access_shadow_credentials.toml
index 6fc57d11382..eb7ad87d754 100644
--- a/rules/windows/credential_access_shadow_credentials.toml
+++ b/rules/windows/credential_access_shadow_credentials.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/01/26"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2023/01/27"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_spn_attribute_modified.toml b/rules/windows/credential_access_spn_attribute_modified.toml
index dda8773c100..5c061a5e65f 100644
--- a/rules/windows/credential_access_spn_attribute_modified.toml
+++ b/rules/windows/credential_access_spn_attribute_modified.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/22"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/credential_access_suspicious_winreg_access_via_sebackup_priv.toml b/rules/windows/credential_access_suspicious_winreg_access_via_sebackup_priv.toml
index f0bba0e9af2..cd8f3a6e50b 100644
--- a/rules/windows/credential_access_suspicious_winreg_access_via_sebackup_priv.toml
+++ b/rules/windows/credential_access_suspicious_winreg_access_via_sebackup_priv.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/16"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/defense_evasion_clearing_windows_security_logs.toml b/rules/windows/defense_evasion_clearing_windows_security_logs.toml
index 6f0500975d8..dfe45a89a5e 100644
--- a/rules/windows/defense_evasion_clearing_windows_security_logs.toml
+++ b/rules/windows/defense_evasion_clearing_windows_security_logs.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/11/12"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic", "Anabella Cristaldi"]
diff --git a/rules/windows/discovery_privileged_localgroup_membership.toml b/rules/windows/discovery_privileged_localgroup_membership.toml
index 6c19cb1b041..e43c35fbe85 100644
--- a/rules/windows/discovery_privileged_localgroup_membership.toml
+++ b/rules/windows/discovery_privileged_localgroup_membership.toml
@@ -1,8 +1,8 @@
 [metadata]
 creation_date = "2020/10/15"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
-updated_date = "2023/01/18"
+updated_date = "2023/02/01"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
 
diff --git a/rules/windows/discovery_whoami_command_activity.toml b/rules/windows/discovery_whoami_command_activity.toml
index a26444bfbc1..8f4b320fb3d 100644
--- a/rules/windows/discovery_whoami_command_activity.toml
+++ b/rules/windows/discovery_whoami_command_activity.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/02/18"
-integration = ["endpoint", "windows"]
+integration = ["endpoint", "system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/14"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/lateral_movement_remote_service_installed_winlog.toml b/rules/windows/lateral_movement_remote_service_installed_winlog.toml
index e5b5e3312d4..b7138c58088 100644
--- a/rules/windows/lateral_movement_remote_service_installed_winlog.toml
+++ b/rules/windows/lateral_movement_remote_service_installed_winlog.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/30"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/lateral_movement_remote_task_creation_winlog.toml b/rules/windows/lateral_movement_remote_task_creation_winlog.toml
index 92a5476d8e5..c9f6ce71031 100644
--- a/rules/windows/lateral_movement_remote_task_creation_winlog.toml
+++ b/rules/windows/lateral_movement_remote_task_creation_winlog.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/lateral_movement_service_control_spawned_script_int.toml b/rules/windows/lateral_movement_service_control_spawned_script_int.toml
index 49131e48745..b8f0b59ea04 100644
--- a/rules/windows/lateral_movement_service_control_spawned_script_int.toml
+++ b/rules/windows/lateral_movement_service_control_spawned_script_int.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2020/02/18"
-integration = ["endpoint", "windows"]
+integration = ["endpoint", "system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_ad_adminsdholder.toml b/rules/windows/persistence_ad_adminsdholder.toml
index a4e95f7f4a0..b980b44ba6a 100644
--- a/rules/windows/persistence_ad_adminsdholder.toml
+++ b/rules/windows/persistence_ad_adminsdholder.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/01/31"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_dontexpirepasswd_account.toml b/rules/windows/persistence_dontexpirepasswd_account.toml
index e4b4fb68da1..e5600b10f26 100644
--- a/rules/windows/persistence_dontexpirepasswd_account.toml
+++ b/rules/windows/persistence_dontexpirepasswd_account.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/22"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_msds_alloweddelegateto_krbtgt.toml b/rules/windows/persistence_msds_alloweddelegateto_krbtgt.toml
index 04bed4403ee..686ef911974 100644
--- a/rules/windows/persistence_msds_alloweddelegateto_krbtgt.toml
+++ b/rules/windows/persistence_msds_alloweddelegateto_krbtgt.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/01/27"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_remote_password_reset.toml b/rules/windows/persistence_remote_password_reset.toml
index 22cb2b45eb0..f729c5ac691 100644
--- a/rules/windows/persistence_remote_password_reset.toml
+++ b/rules/windows/persistence_remote_password_reset.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/10/18"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_scheduled_task_creation_winlog.toml b/rules/windows/persistence_scheduled_task_creation_winlog.toml
index bfbfc72f3e0..ff0e183f963 100644
--- a/rules/windows/persistence_scheduled_task_creation_winlog.toml
+++ b/rules/windows/persistence_scheduled_task_creation_winlog.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_scheduled_task_updated.toml b/rules/windows/persistence_scheduled_task_updated.toml
index 84e5f0710eb..e7b247eac9b 100644
--- a/rules/windows/persistence_scheduled_task_updated.toml
+++ b/rules/windows/persistence_scheduled_task_updated.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_sdprop_exclusion_dsheuristics.toml b/rules/windows/persistence_sdprop_exclusion_dsheuristics.toml
index 91abd2570b5..57c6db82c6a 100644
--- a/rules/windows/persistence_sdprop_exclusion_dsheuristics.toml
+++ b/rules/windows/persistence_sdprop_exclusion_dsheuristics.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/24"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_service_windows_service_winlog.toml b/rules/windows/persistence_service_windows_service_winlog.toml
index 697f335e2b2..934ed8f19a0 100644
--- a/rules/windows/persistence_service_windows_service_winlog.toml
+++ b/rules/windows/persistence_service_windows_service_winlog.toml
@@ -1,6 +1,6 @@
 [metadata]
 creation_date = "2022/08/30"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
diff --git a/rules/windows/persistence_temp_scheduled_task.toml b/rules/windows/persistence_temp_scheduled_task.toml
index de92808fba1..fa1ce3ad378 100644
--- a/rules/windows/persistence_temp_scheduled_task.toml
+++ b/rules/windows/persistence_temp_scheduled_task.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/29"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/persistence_user_account_added_to_privileged_group_ad.toml b/rules/windows/persistence_user_account_added_to_privileged_group_ad.toml
index 104305e2d56..805b4dab5ef 100644
--- a/rules/windows/persistence_user_account_added_to_privileged_group_ad.toml
+++ b/rules/windows/persistence_user_account_added_to_privileged_group_ad.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/01/09"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic", "Skoetting"]
diff --git a/rules/windows/persistence_user_account_creation_event_logs.toml b/rules/windows/persistence_user_account_creation_event_logs.toml
index b4334f7b438..796dfc59741 100644
--- a/rules/windows/persistence_user_account_creation_event_logs.toml
+++ b/rules/windows/persistence_user_account_creation_event_logs.toml
@@ -1,8 +1,8 @@
 [metadata]
 creation_date = "2021/01/04"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "development"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Skoetting"]
diff --git a/rules/windows/privilege_escalation_create_process_as_different_user.toml b/rules/windows/privilege_escalation_create_process_as_different_user.toml
index f43fedaa162..3ba9885118d 100644
--- a/rules/windows/privilege_escalation_create_process_as_different_user.toml
+++ b/rules/windows/privilege_escalation_create_process_as_different_user.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/08/30"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_credroaming_ldap.toml b/rules/windows/privilege_escalation_credroaming_ldap.toml
index 4dfcff827b0..8c14dd73dd3 100644
--- a/rules/windows/privilege_escalation_credroaming_ldap.toml
+++ b/rules/windows/privilege_escalation_credroaming_ldap.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/11/09"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_group_policy_iniscript.toml b/rules/windows/privilege_escalation_group_policy_iniscript.toml
index f07dc17f2e3..866b4b82c85 100644
--- a/rules/windows/privilege_escalation_group_policy_iniscript.toml
+++ b/rules/windows/privilege_escalation_group_policy_iniscript.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/11/08"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
@@ -18,7 +18,7 @@ note = """## Triage and analysis
 
 ### Investigating Scheduled Task Execution at Scale via GPO
 
-Group Policy Objects (GPOs) can be used by attackers to instruct arbitrarily large groups of clients to execute specified commands at startup, logon, shutdown, and logoff. This is done by creating or modifying the `scripts.ini` or `psscripts.ini` files. The scripts are stored in the following paths: 
+Group Policy Objects (GPOs) can be used by attackers to instruct arbitrarily large groups of clients to execute specified commands at startup, logon, shutdown, and logoff. This is done by creating or modifying the `scripts.ini` or `psscripts.ini` files. The scripts are stored in the following paths:
   - `<GPOPath>\\Machine\\Scripts\\`
   - `<GPOPath>\\User\\Scripts\\`
 
diff --git a/rules/windows/privilege_escalation_group_policy_privileged_groups.toml b/rules/windows/privilege_escalation_group_policy_privileged_groups.toml
index 89466e86ee8..43162f135bd 100644
--- a/rules/windows/privilege_escalation_group_policy_privileged_groups.toml
+++ b/rules/windows/privilege_escalation_group_policy_privileged_groups.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/11/08"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_group_policy_scheduled_task.toml b/rules/windows/privilege_escalation_group_policy_scheduled_task.toml
index ea70c44f18b..b7b2f2195bc 100644
--- a/rules/windows/privilege_escalation_group_policy_scheduled_task.toml
+++ b/rules/windows/privilege_escalation_group_policy_scheduled_task.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/11/08"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_krbrelayup_service_creation.toml b/rules/windows/privilege_escalation_krbrelayup_service_creation.toml
index b37a2f48a69..a515d0ef53d 100644
--- a/rules/windows/privilege_escalation_krbrelayup_service_creation.toml
+++ b/rules/windows/privilege_escalation_krbrelayup_service_creation.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/04/27"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_samaccountname_spoofing_attack.toml b/rules/windows/privilege_escalation_samaccountname_spoofing_attack.toml
index b0e7096a7c1..fdf9d814211 100644
--- a/rules/windows/privilege_escalation_samaccountname_spoofing_attack.toml
+++ b/rules/windows/privilege_escalation_samaccountname_spoofing_attack.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2021/12/12"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_suspicious_dnshostname_update.toml b/rules/windows/privilege_escalation_suspicious_dnshostname_update.toml
index 1b615e0836b..34b89a37171 100644
--- a/rules/windows/privilege_escalation_suspicious_dnshostname_update.toml
+++ b/rules/windows/privilege_escalation_suspicious_dnshostname_update.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/05/11"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/rules/windows/privilege_escalation_windows_service_via_unusual_client.toml b/rules/windows/privilege_escalation_windows_service_via_unusual_client.toml
index b21d26b5b54..135cc0bf65e 100644
--- a/rules/windows/privilege_escalation_windows_service_via_unusual_client.toml
+++ b/rules/windows/privilege_escalation_windows_service_via_unusual_client.toml
@@ -1,10 +1,10 @@
 [metadata]
 creation_date = "2022/02/07"
-integration = ["windows"]
+integration = ["system", "windows"]
 maturity = "production"
 min_stack_comments = "New fields added: required_fields, related_integrations, setup"
 min_stack_version = "8.3.0"
-updated_date = "2022/12/21"
+updated_date = "2023/02/01"
 
 [rule]
 author = ["Elastic"]
diff --git a/tests/test_all_rules.py b/tests/test_all_rules.py
index 48c36ae7a00..3b59020d1aa 100644
--- a/tests/test_all_rules.py
+++ b/tests/test_all_rules.py
@@ -443,36 +443,43 @@ def test_deprecated_rules(self):
     def test_integration_tag(self):
         """Test integration rules defined by metadata tag."""
         failures = []
-        non_dataset_packages = ["apm", "endpoint", "windows", "winlog"]
+        non_dataset_packages = definitions.NON_DATASET_PACKAGES + ["winlog"]
 
         packages_manifest = load_integrations_manifests()
         valid_integration_folders = [p.name for p in list(Path(INTEGRATION_RULE_DIR).glob("*")) if p.name != 'endpoint']
 
         for rule in self.production_rules:
-            rule_integrations = rule.contents.metadata.get('integration')
-            if rule_integrations:
+            if isinstance(rule.contents.data, QueryRuleData) and rule.contents.data.language != 'lucene':
+                rule_integrations = rule.contents.metadata.get('integration') or []
                 rule_integrations = [rule_integrations] if isinstance(rule_integrations, str) else rule_integrations
+                data = rule.contents.data
+                meta = rule.contents.metadata
+                package_integrations = TOMLRuleContents.get_packaged_integrations(data, meta, packages_manifest)
+                package_integrations_list = list(set([integration["package"] for integration in package_integrations]))
+                indices = data.get('index')
                 for rule_integration in rule_integrations:
-                    # checks if metadata tag matches from a list of integrations in EPR
-                    if rule_integration not in packages_manifest.keys():
-                        err_msg = f"{self.rule_str(rule)} integration '{rule_integration}' unknown"
-                        failures.append(err_msg)
 
                     # checks if the rule path matches the intended integration
                     if rule_integration in valid_integration_folders:
-                        if rule_integration != rule.path.parent.name:
+                        if rule.path.parent.name not in rule_integrations:
                             err_msg = f'{self.rule_str(rule)} {rule_integration} tag, path is {rule.path.parent.name}'
                             failures.append(err_msg)
 
-            else:
-                # checks if event.dataset exists in query object and a tag exists in metadata
-                if isinstance(rule.contents.data, QueryRuleData) and rule.contents.data.language != 'lucene':
-                    trc = TOMLRuleContents(rule.contents.metadata, rule.contents.data)
-                    package_integrations = trc._get_packaged_integrations(packages_manifest)
-                    if package_integrations:
-                        err_msg = f'{self.rule_str(rule)} integration tag should exist: '
+                    # checks if an index pattern exists if the package integration tag exists
+                    integration_string = "|".join(indices)
+                    if not re.search(rule_integration, integration_string):
+                        if rule_integration == "windows" and re.search("winlog", integration_string):
+                            continue
+                        err_msg = f'{self.rule_str(rule)} {rule_integration} tag, index pattern missing.'
                         failures.append(err_msg)
 
+                # checks if event.dataset exists in query object and a tag exists in metadata
+                # checks if metadata tag matches from a list of integrations in EPR
+                if package_integrations and sorted(rule_integrations) != sorted(package_integrations_list):
+                    err_msg = f'{self.rule_str(rule)} integration tags: {rule_integrations} != ' \
+                              f'package integrations: {package_integrations_list}'
+                    failures.append(err_msg)
+                else:
                     # checks if rule has index pattern integration and the integration tag exists
                     # ignore the External Alerts rule, Threat Indicator Matching Rules, Guided onboarding
                     ignore_ids = [