diff --git a/fixlib/fixlib/baseresources.py b/fixlib/fixlib/baseresources.py index d18d0797a3..411ec1e4fa 100644 --- a/fixlib/fixlib/baseresources.py +++ b/fixlib/fixlib/baseresources.py @@ -4,23 +4,22 @@ import hashlib import weakref from abc import ABC, abstractmethod +from collections import defaultdict from copy import deepcopy -from datetime import datetime, timezone, timedelta +from datetime import datetime, timedelta from enum import Enum, StrEnum, unique from functools import wraps, cached_property from typing import Dict, Iterator, List, ClassVar, Optional, TypedDict, Any, TypeVar, Type, Callable, Set, Tuple -from collections import defaultdict from attr import resolve_types from attrs import define, field, Factory, frozen, evolve from prometheus_client import Counter, Summary +from fixlib.basecategories import Category from fixlib.json import from_json as _from_json, to_json as _to_json, to_json_str from fixlib.logger import log from fixlib.types import Json from fixlib.utils import make_valid_timestamp, utc_str, utc -from fixlib.basecategories import Category - metrics_resource_pre_cleanup_exceptions = Counter( "resource_pre_cleanup_exceptions_total", @@ -246,6 +245,32 @@ def __str__(self) -> str: MetricNameWithUnit = str +class Severity(StrEnum): + info = "info" + low = "low" + medium = "medium" + high = "high" + critical = "critical" + + +@define(slots=True) +class Finding: + title: str + severity: Severity = Severity.medium + description: Optional[str] = None + remediation: Optional[str] = None + created_at: Optional[datetime] = None + details: Optional[Json] = None + + +@define(slots=True) +class Assessment: + # The provider of the security assessment + provider: str + # All findings of the security provider to this resource + findings: List[Finding] = field(factory=list) + + @define(eq=False, slots=False, kw_only=True) class BaseResource(ABC): """ @@ -305,6 +330,8 @@ class BaseResource(ABC): _resource_usage: Dict[MetricNameWithUnit, Dict[str, float]] = field(factory=lambda: defaultdict(dict)) # Deep link into the cloud provider's console _provider_link: Optional[str] = None + # Assessment details for this resource: multiple providers can append their findings + _assessments: List[Assessment] = field(factory=list) ctime: Optional[datetime] = field( default=None, diff --git a/fixlib/fixlib/core/model_export.py b/fixlib/fixlib/core/model_export.py index 2f6821c9f5..53196bca45 100644 --- a/fixlib/fixlib/core/model_export.py +++ b/fixlib/fixlib/core/model_export.py @@ -12,7 +12,7 @@ from attrs import Attribute from fixlib.baseresources import BaseResource -from fixlib.json import from_json +from fixlib.json import from_json, to_json from fixlib.types import Json from fixlib.utils import type_str @@ -85,6 +85,8 @@ def check(to_check: type) -> None: for subclass in clazz.__subclasses__(): check(subclass) for field in attrs.fields(clazz): + if field.name.startswith("_"): # ignore private properties + continue check(field.type) elif is_enum(clazz): all_classes.add(clazz) @@ -349,6 +351,8 @@ def node_to_dict(node: BaseResource, changes_only: bool = False, include_revisio metadata["protected"] = True if link := node._provider_link: metadata["provider_link"] = link + if assessments := node._assessments: + metadata["assessments"] = to_json(assessments) node_dict["reported"] = get_node_attributes(node) node_dict["metadata"] = metadata