[feat][resotocore] Allow loading synchronized benchmarks (#1775)

resotocore/resotocore/cli/command.py

@@ -4786,6 +4786,11 @@ class ReportCommand(CLICommand, EntityProvider):
                          [--only-check-results]
                          [--sync-security-section]
                          [--run-id <run-id>]
+    report benchmark load <benchmark-id>
+                         [--accounts <account-id>]
+                         [--severity <level>]
+                         [--only-failing]
+                         [--only-check-results]
     report checks list
     report checks show <check-id>
     report checks run <check-id>
@@ -4851,6 +4856,10 @@ class ReportCommand(CLICommand, EntityProvider):
     > report benchmark run aws_cis_1_5
       ... output suppressed ...
 
+    # Load the benchmark from a previous run with sync enabled
+    > report benchmark load aws_cis_1_5
+      ... output suppressed ...
+
     # Run benchmark against account 111 and 222, check only critical checks
     # The resulting report should contain only failed checks.
     > report benchmark run aws_cis_1_5 --accounts 111 222  --severity critical --only-failing
@@ -4894,7 +4903,7 @@ def info(self) -> str:
 
     @staticmethod
     def is_run_action(arg: Optional[str]) -> bool:
-        return ReportCommand.action_from_arg(arg) in ("benchmark_run", "check_run")
+        return ReportCommand.action_from_arg(arg) in ("benchmark_run", "check_run", "benchmark_load")
 
     @staticmethod
     def action_from_arg(arg: Optional[str]) -> Optional[str]:
@@ -4905,6 +4914,8 @@ def action_from_arg(arg: Optional[str]) -> Optional[str]:
             return "benchmark_show"
         elif len(args) >= 3 and args[0] in ("benchmark", "benchmarks") and args[1] == "run":
             return "benchmark_run"
+        elif len(args) >= 3 and args[0] in ("benchmark", "benchmarks") and args[1] == "load":
+            return "benchmark_load"
         elif len(args) == 2 and args[0] in ("check", "checks") and args[1] == "list":
             return "check_list"
         elif len(args) == 3 and args[0] in ("check", "checks") and args[1] == "show":
@@ -4947,6 +4958,19 @@ async def run_benchmark(parsed_args: Namespace) -> AsyncIterator[Json]:
                     for node in result.to_graph(parsed_args.only_check_results):
                         yield node
 
+        async def load_benchmark(parsed_args: Namespace) -> AsyncIterator[Json]:
+            results = await self.dependencies.inspector.load_benchmarks(
+                ctx.graph_name,
+                benchmark_names=parsed_args.identifier,
+                accounts=parsed_args.accounts,
+                severity=parsed_args.severity,
+                only_failing=parsed_args.only_failing,
+            )
+            for result in results.values():
+                if not result.is_empty():
+                    for node in result.to_graph(parsed_args.only_check_results):
+                        yield node
+
         async def run_check(parsed_args: Namespace) -> AsyncIterator[Json]:
             result = await self.dependencies.inspector.perform_checks(
                 ctx.graph_name,
@@ -4980,6 +5004,9 @@ async def show_help() -> AsyncIterator[str]:
         elif action == "benchmark_run":
             parsed = run_parser.parse_args(args[2].split() if len(args) > 2 else [])
             return CLISource.no_count(partial(run_benchmark, parsed), required_permissions={Permission.read})
+        elif action == "benchmark_load":
+            parsed = run_parser.parse_args(args[2].split() if len(args) > 2 else [])
+            return CLISource.no_count(partial(load_benchmark, parsed), required_permissions={Permission.read})
         elif action == "check_list":
             return CLISource.no_count(list_checks, required_permissions={Permission.read})
         elif action == "check_show":

resotocore/resotocore/config/config_handler_service.py

@@ -78,7 +78,7 @@ async def coerce_and_check_model(self, cfg_id: ConfigId, config: Json, validate:
         if validator:
             validation = await self.validation_db.get(validator)
             if validation and validation.external_validation and validate:
-                await self.acknowledge_config_change(validator, final_config)
+                await self.acknowledge_config_change(validator, cfg_id, final_config)
 
         # If we come here, everything is fine
         return final_config
@@ -239,7 +239,7 @@ def mkstr(val: Any) -> str:
         else:
             return None
 
-    async def acknowledge_config_change(self, cfg_id: str, config: Json) -> None:
+    async def acknowledge_config_change(self, validator: str, cfg_id: ConfigId, config: Json) -> None:
         """
         In case an external entity should acknowledge this config change.
         This method either return, which signals success or throws an exception.
@@ -248,8 +248,8 @@ async def acknowledge_config_change(self, cfg_id: str, config: Json) -> None:
         task = WorkerTask(
             TaskId(uuid_str()),
             WorkerTaskName.validate_config,
-            {"config_id": cfg_id},
-            {"task": WorkerTaskName.validate_config, "config": config},
+            {"config_id": validator},
+            {"task": WorkerTaskName.validate_config, "config": config, "config_id": cfg_id},
             future,
             timedelta(seconds=30),
         )

resotocore/resotocore/config/core_config_handler.py

@@ -108,7 +108,7 @@ def validate_snapshot_schedule() -> Optional[Json]:
         elif CheckConfigRoot in holder:
             return await self.inspector.validate_check_collection_config(task_data["config"])
         elif BenchmarkConfigRoot in holder:
-            return await self.inspector.validate_benchmark_config(task_data["config"])
+            return await self.inspector.validate_benchmark_config(task_data["config_id"], task_data["config"])
         elif ResotoCoreSnapshotsRoot in holder:
             return validate_snapshot_schedule()
         else:

resotocore/resotocore/report/__init__.py

@@ -356,6 +356,18 @@ async def list_checks(
         :return: the list of matching checks
         """
 
+    @abstractmethod
+    async def load_benchmarks(
+        self,
+        graph: GraphName,
+        benchmark_names: List[str],
+        *,
+        accounts: Optional[List[str]] = None,
+        severity: Optional[ReportSeverity] = None,
+        only_failing: bool = False,
+    ) -> Dict[str, BenchmarkResult]:
+        pass
+
     @abstractmethod
     async def perform_benchmarks(
         self,
@@ -417,7 +429,7 @@ async def list_failing_resources(
         pass
 
     @abstractmethod
-    async def validate_benchmark_config(self, json: Json) -> Optional[Json]:
+    async def validate_benchmark_config(self, cfg_id: ConfigId, json: Json) -> Optional[Json]:
         pass
 
     @abstractmethod

resotocore/resotocore/report/inspector_service.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from collections import defaultdict
-from typing import Optional, List, Dict, Tuple, Callable, AsyncIterator
+from typing import Optional, List, Dict, Tuple, Callable, AsyncIterator, cast
 
 from aiostream import stream
 from attr import define
@@ -14,7 +14,7 @@
 from resotocore.ids import ConfigId, GraphName, NodeId
 from resotocore.model.model import Model
 from resotocore.model.resolve_in_graph import NodePath
-from resotocore.query.model import Query, P
+from resotocore.query.model import Query, P, Term
 from resotocore.report import (
     Inspector,
     ReportCheck,
@@ -36,7 +36,7 @@
 from resotocore.report.report_config import ReportCheckCollectionConfig, BenchmarkConfig
 from resotocore.service import Service
 from resotocore.types import Json
-from resotocore.util import value_in_path, uuid_str
+from resotocore.util import value_in_path, uuid_str, value_in_path_get
 from resotolib.json_bender import Bender, S, bend
 
 log = logging.getLogger(__name__)
@@ -59,6 +59,9 @@ class CheckContext:
     only_failed: bool = False
     parallel_checks: int = 10
 
+    def severities_including(self, severity: ReportSeverity) -> List[ReportSeverity]:
+        return [s for s in ReportSeverity if self.includes_severity(severity)]
+
     def includes_severity(self, severity: ReportSeverity) -> bool:
         if self.severity is None:
             return True
@@ -144,6 +147,46 @@ def inspection_matches(inspection: ReportCheck) -> bool:
 
         return await self.filter_checks(inspection_matches)
 
+    async def load_benchmarks(
+        self,
+        graph: GraphName,
+        benchmark_names: List[str],
+        *,
+        accounts: Optional[List[str]] = None,
+        severity: Optional[ReportSeverity] = None,
+        only_failing: bool = False,
+    ) -> Dict[str, BenchmarkResult]:
+        context = CheckContext(accounts=accounts, severity=severity, only_failed=only_failing)
+        # create query
+        term: Term = P("benchmark").is_in(benchmark_names)
+        if severity:
+            term = term & P("severity").is_in(context.severities_including(severity))
+        term = P.context("security.issues[]", term)
+        if accounts:
+            term = term & P("ancestors.account.reported.id").is_in(accounts)
+        term = term & P("security.has_issues").eq(True)
+        model = QueryModel(Query.by(term), await self.model_handler.load_model(graph))
+
+        # collect all checks
+        benchmarks = {name: await self.__benchmark(benchmark_id(name)) for name in benchmark_names}
+        check_ids = {check for b in benchmarks.values() for check in b.nested_checks()}
+        checks = await self.list_checks(check_ids=list(check_ids), context=context)
+        check_lookup = {check.id: check for check in checks}
+
+        # perform query, map resources and create lookup map
+        check_results: Dict[str, SingleCheckResult] = defaultdict(lambda: defaultdict(list))
+        async with await self.db_access.get_graph_db(graph).search_list(model) as cursor:
+            async for entry in cursor:
+                if account_id := value_in_path(entry, NodePath.ancestor_account_id):
+                    mapped = bend(ReportResourceData, entry)
+                    for issue in value_in_path_get(entry, NodePath.security_issues, cast(List[Json], [])):
+                        if check := issue.get("check"):
+                            check_results[check][account_id].append(mapped)
+        return {
+            name: self.__to_result(benchmark, check_lookup, check_results, context)
+            for name, benchmark in benchmarks.items()
+        }
+
     async def perform_benchmarks(
         self,
         graph: GraphName,
@@ -365,9 +408,12 @@ async def __list_accounts(self, benchmark: Benchmark, graph: GraphName) -> List[
             ids = [value_in_path(a, NodePath.reported_id) async for a in crs]
             return [aid for aid in ids if aid is not None]
 
-    async def validate_benchmark_config(self, json: Json) -> Optional[Json]:
+    async def validate_benchmark_config(self, cfg_id: ConfigId, json: Json) -> Optional[Json]:
         try:
             benchmark = BenchmarkConfig.from_config(ConfigEntity(ResotoReportBenchmark, json))
+            bid = cfg_id.rsplit(".", 1)[-1]
+            if benchmark.id != bid:
+                return {"error": f"Benchmark id should be {bid} (same as the config name). Got {benchmark.id}"}
             all_checks = {c.id for c in await self.filter_checks()}
             missing = []
             for check in benchmark.nested_checks():

resotocore/resotocore/static/api-doc.yaml

@@ -2752,6 +2752,17 @@ paths:
           required: false
           explode: false
           example: 123456789012,123456789013
+        - name: action
+          in: query
+          description: |
+            The action to perform. You can either run (create) or load the benchmark.
+            Loading the benchmark requires, that a benchmark has been synchronized to the database before.
+            In case the resources haven't changed since the last synchronization, the result is the same,
+            but the load action is much faster.
+          schema:
+            type: string
+            enum: [run, load]
+            default: run
       responses:
         "200":
           description: "The checks result."

resotocore/resotocore/web/api.py

@@ -629,7 +629,13 @@ async def perform_benchmark(self, request: Request, deps: TenantDependencies) ->
         graph = GraphName(request.match_info["graph_id"])
         acc = request.query.get("accounts")
         accounts = [a.strip() for a in acc.split(",")] if acc else None
-        results = await deps.inspector.perform_benchmarks(graph, [benchmark], accounts=accounts)
+        action = request.query.get("action", "run")
+        if action == "run":
+            results = await deps.inspector.perform_benchmarks(graph, [benchmark], accounts=accounts)
+        elif action == "load":
+            results = await deps.inspector.load_benchmarks(graph, [benchmark], accounts=accounts)
+        else:
+            raise ValueError(f"Unknown action {action}. One of run or load is expected.")
         result_graph = results[benchmark].to_graph()
         async with stream.iterate(result_graph).stream() as streamer:
             return await self.stream_response_from_gen(request, streamer, len(result_graph))

resotocore/tests/resotocore/cli/command_test.py

@@ -1059,9 +1059,11 @@ async def execute(cmd: str, _: Type[T]) -> List[T]:
     # without output transformer, a markdown report is generated
     assert len((await execute("report check run test_test_some_check", str))) == 1
     # execute the test benchmark
-    assert len((await execute("report benchmark run test | dump", Json))) == 9
+    assert len((await execute("report benchmark run test --sync-security-section | dump", Json))) == 9
     assert len((await execute("report benchmark run test --only-failing | dump", Json))) == 9
     assert len((await execute("report benchmark run test --severity critical | dump", Json))) == 0
+    # load the benchmark from the last sync
+    assert len((await execute("report benchmark load test | dump", Json))) == 9
 
 
 @pytest.mark.asyncio

resotocore/tests/resotocore/config/core_config_handler_test.py

@@ -14,6 +14,8 @@
     CustomCommandsConfig,
     ResotoCoreSnapshotsRoot,
 )
+from resotocore.model.typed_model import to_js
+from resotocore.report import BenchmarkConfigRoot, Benchmark
 from resotocore.system_start import empty_config
 from resotocore.message_bus import MessageBus, CoreMessage
 from resotocore.ids import ConfigId
@@ -62,7 +64,7 @@ async def callback(config_id: ConfigId) -> None:
 
 
 @mark.asyncio
-async def test_validation(core_config_handler: CoreConfigHandler) -> None:
+async def test_validation(core_config_handler: CoreConfigHandler, test_benchmark: Benchmark) -> None:
     validate = core_config_handler.validate_config_entry
 
     # empty config is valid
@@ -93,6 +95,10 @@ async def test_validation(core_config_handler: CoreConfigHandler) -> None:
         await validate({"config": {ResotoCoreSnapshotsRoot: {"foo-label": {"schedule": "foo bar", "retain": 42}}}})
         is not None
     )
+    # make sure that the benchmark id and config_id are the same
+    assert await validate({"config": {BenchmarkConfigRoot: {"id": "some"}}, "config_id": "some_other"}) is not None
+    # a valid benchmark config passes the check
+    assert await validate({"config": {BenchmarkConfigRoot: to_js(test_benchmark)}, "config_id": "test"}) is None
 
 
 @mark.asyncio

resotocore/tests/resotocore/report/inspector_service_test.py

@@ -1,11 +1,13 @@
+from typing import Dict
+
 from pytest import fixture
 
 from resotocore.cli.cli import CLIService
 from resotocore.config import ConfigEntity
 from resotocore.db.model import QueryModel
 from resotocore.ids import ConfigId, GraphName
 from resotocore.query.model import P, Query
-from resotocore.report import BenchmarkConfigRoot, CheckConfigRoot
+from resotocore.report import BenchmarkConfigRoot, CheckConfigRoot, BenchmarkResult
 from resotocore.report.inspector_service import InspectorService, check_id, benchmark_id
 from resotocore.report.report_config import (
     config_model,
@@ -66,7 +68,7 @@ def benchmark() -> Json:
         report_benchmark=dict(
             title="test_benchmark",
             description="test_benchmark",
-            id="test_benchmark",
+            id="test",
             framework="test",
             version="1.0",
             checks=["test_test_search", "test_test_cmd"],
@@ -104,23 +106,26 @@ async def test_list_inspect_checks(inspector_service: InspectorService) -> None:
 
 
 async def test_perform_benchmark(inspector_service_with_test_benchmark: InspectorService) -> None:
+    def assert_result(results: Dict[str, BenchmarkResult]) -> None:
+        result = results["test"]
+        assert result.checks[0].number_of_resources_failing == 10
+        assert result.checks[1].number_of_resources_failing == 10
+        filtered = result.filter_result(filter_failed=True)
+        assert filtered.checks[0].number_of_resources_failing == 10
+        assert len(filtered.checks[0].resources_failing_by_account["sub_root"]) == 10
+        assert filtered.checks[1].number_of_resources_failing == 10
+        assert len(filtered.checks[1].resources_failing_by_account["sub_root"]) == 10
+        passing, failing = result.passing_failing_checks_for_account("sub_root")
+        assert len(passing) == 0
+        assert len(failing) == 2
+        passing, failing = result.passing_failing_checks_for_account("does_not_exist")
+        assert len(passing) == 2
+        assert len(failing) == 0
+
     inspector = inspector_service_with_test_benchmark
     graph_name = GraphName(inspector.cli.env["graph"])
-    results = await inspector.perform_benchmarks(graph_name, ["test"], sync_security_section=True)
-    result = results["test"]
-    assert result.checks[0].number_of_resources_failing == 10
-    assert result.checks[1].number_of_resources_failing == 10
-    filtered = result.filter_result(filter_failed=True)
-    assert filtered.checks[0].number_of_resources_failing == 10
-    assert len(filtered.checks[0].resources_failing_by_account["sub_root"]) == 10
-    assert filtered.checks[1].number_of_resources_failing == 10
-    assert len(filtered.checks[1].resources_failing_by_account["sub_root"]) == 10
-    passing, failing = result.passing_failing_checks_for_account("sub_root")
-    assert len(passing) == 0
-    assert len(failing) == 2
-    passing, failing = result.passing_failing_checks_for_account("does_not_exist")
-    assert len(passing) == 2
-    assert len(failing) == 0
+    performed = await inspector.perform_benchmarks(graph_name, ["test"], sync_security_section=True)
+    assert_result(performed)
 
     # make sure the result is persisted as part of the node
     async def count_vulnerable() -> int:
@@ -132,6 +137,10 @@ async def count_vulnerable() -> int:
 
     assert await count_vulnerable() == 10
 
+    # loading the result from the db should give the same information
+    loaded = await inspector.load_benchmarks(graph_name, ["test"])
+    assert_result(loaded)
+
 
 async def test_benchmark_node_result(inspector_service_with_test_benchmark: InspectorService) -> None:
     inspector = inspector_service_with_test_benchmark
@@ -158,8 +167,9 @@ async def test_predefined_benchmarks(inspector_service: InspectorService) -> Non
     assert len(benchmarks) > 0
     for name, check in benchmarks.items():
         config = {BenchmarkConfigRoot: check}
-        assert (await inspector_service.validate_benchmark_config(config)) is None
-        benchmark = BenchmarkConfig.from_config(ConfigEntity(ConfigId("test"), config))
+        cfg_id = ConfigId(name)
+        assert (await inspector_service.validate_benchmark_config(cfg_id, config)) is None
+        benchmark = BenchmarkConfig.from_config(ConfigEntity(cfg_id, config))
         assert benchmark.clouds == ["aws"]