diff --git a/HolmSecurity/CHANGELOG.md b/HolmSecurity/CHANGELOG.md index 67f732c3f..aa9356156 100644 --- a/HolmSecurity/CHANGELOG.md +++ b/HolmSecurity/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [1.1.0] - 2026-07-20 + +### Added + +- Vulnerability asset connector, correlates them with managed devices from, and maps them to the OCSF Vulnerability Finding model. + +### Changed + +- Account validator updates to reflect vulnerability setup + ## [1.0.0] - 2026-07-17 ### Added diff --git a/HolmSecurity/connector_holm_security_vulnerability_assets.json b/HolmSecurity/connector_holm_security_vulnerability_assets.json new file mode 100644 index 000000000..ab03022a7 --- /dev/null +++ b/HolmSecurity/connector_holm_security_vulnerability_assets.json @@ -0,0 +1,37 @@ +{ + "name": "Holm Security Vulnerabilities", + "description": "Fetch Holm Security vulnerability findings and correlate them with managed devices", + "uuid": "f1ccabe4-8193-492a-852f-c32f6479d28b", + "docker_parameters": "holm_security_vulnerability_asset_connector", + "type": "asset", + "capability": "vulnerability_enrichment", + "arguments": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "sekoia_base_url": { + "description": "Sekoia base URL", + "type": "string" + }, + "frequency": { + "type": "integer", + "description": "Batch frequency in seconds", + "minimum": 1, + "default": 60 + }, + "sekoia_api_key": { + "description": "API key to use from sekoia.io", + "type": "string" + }, + "batch_size": { + "type": "integer", + "description": "Number of assets to process in one batch", + "minimum": 1, + "default": 100 + } + }, + "required": ["sekoia_api_key"], + "secrets": ["sekoia_api_key"], + "internal": ["sekoia_base_url", "sekoia_api_key", "frequency", "batch_size"] + } +} diff --git a/HolmSecurity/holm_security/account_validator.py b/HolmSecurity/holm_security/account_validator.py index 218ec9db6..1d9c800a4 100644 --- a/HolmSecurity/holm_security/account_validator.py +++ b/HolmSecurity/holm_security/account_validator.py @@ -8,13 +8,18 @@ class HolmSecurityAccountValidator(AccountValidator): - """Validate Holm Security credentials by pinging the devices endpoint. + """Validate Holm Security credentials against the endpoints used by the connectors. - A ``200 OK`` on ``GET /v2/devices?page_size=1`` confirms the token is valid. + The token is confirmed valid only when both ``GET /v2/devices`` and + ``GET /v2/net-assets/report/vulnerabilities/`` return ``200 OK``. """ TIMEOUT = 30 - VALIDATION_ENDPOINT = "/v2/devices" + # (endpoint, query parameters) pairs that must each return 200. + VALIDATION_ENDPOINTS: list[tuple[str, dict[str, int]]] = [ + ("/v2/devices", {"page_size": 1}), + ("/v2/net-assets/report/vulnerabilities/", {"limit": 1}), + ] @cached_property def base_url(self) -> str: @@ -24,12 +29,8 @@ def base_url(self) -> str: def client(self) -> ApiClient: return ApiClient(base_url=self.base_url, token=self.module.configuration["api_token"]) - @cached_property - def validation_url(self) -> str: - return f"{self.base_url}{self.VALIDATION_ENDPOINT}" - @staticmethod - def _describe_error(response: Response) -> str: + def _describe_error(url: str, response: Response) -> str: code = response.status_code try: @@ -38,45 +39,47 @@ def _describe_error(response: Response) -> str: detail = response.text if 400 <= code < 500: - return f"Client error ({code}) while validating Holm Security credentials: {detail}" + return f"Client error ({code}) while validating Holm Security credentials at {url}: {detail}" if 500 <= code < 600: - return f"Server error ({code}) while validating Holm Security credentials: {detail}" - return f"Unexpected status ({code}) while validating Holm Security credentials: {detail}" + return f"Server error ({code}) while validating Holm Security credentials at {url}: {detail}" + return f"Unexpected status ({code}) while validating Holm Security credentials at {url}: {detail}" - def validate(self) -> bool: - self.log(message="Starting credentials validation for Holm Security asset connector", level="info") + def _check_endpoint(self, endpoint: str, params: dict[str, int]) -> bool: + url = f"{self.base_url}{endpoint}" try: - response = self.client.get(self.validation_url, params={"page_size": 1}, timeout=self.TIMEOUT) + response = self.client.get(url, params=params, timeout=self.TIMEOUT) except Timeout: - message = f"Timeout while validating Holm Security credentials at {self.validation_url}" + message = f"Timeout while validating Holm Security credentials at {url}" self.log(message=message, level="error") self.error(message) - return False - except RequestException as error: - message = f"Network error while validating Holm Security credentials: {error}" + message = f"Network error while validating Holm Security credentials at {url}: {error}" self.log(message=message, level="error") self.error(message) - return False - except Exception as error: - message = f"Unexpected error while validating Holm Security credentials: {error}" + message = f"Unexpected error while validating Holm Security credentials at {url}: {error}" self.log(message=message, level="error") self.error(message) - return False if response.status_code == 200: - self.log(message="Holm Security credentials validated successfully", level="info") return True - message = self._describe_error(response) - + message = self._describe_error(url, response) self.log(message=message, level="error") self.error(message) - return False + + def validate(self) -> bool: + self.log(message="Starting credentials validation for Holm Security asset connector", level="info") + + for endpoint, params in self.VALIDATION_ENDPOINTS: + if not self._check_endpoint(endpoint, params): + return False + + self.log(message="Holm Security credentials validated successfully", level="info") + return True diff --git a/HolmSecurity/holm_security/asset_connector/device_assets.py b/HolmSecurity/holm_security/asset_connector/device_assets.py index d0552cb7d..ad759876d 100644 --- a/HolmSecurity/holm_security/asset_connector/device_assets.py +++ b/HolmSecurity/holm_security/asset_connector/device_assets.py @@ -13,14 +13,11 @@ DeviceTypeId, DeviceTypeStr, NetworkInterface, - NetworkInterfaceTypeId, - NetworkInterfaceTypeStr, OperatingSystem, - OSTypeId, - OSTypeStr, ) from sekoia_automation.storage import PersistentJSON +from holm_security.asset_connector import mappers from holm_security.asset_connector.models import HolmDevice, HolmDevicePage from holm_security.client import ApiClient @@ -48,16 +45,6 @@ class HolmSecurityDeviceAssetConnector(AssetConnector): TYPE_NAME: str = "Device Inventory Info: Collect" TYPE_UID: int = 500102 - # Holm os_family -> OCSF OS type mapping - OS_FAMILY_MAP: dict[str, tuple[OSTypeStr, OSTypeId]] = { - "windows": (OSTypeStr.WINDOWS, OSTypeId.WINDOWS), - "linux": (OSTypeStr.LINUX, OSTypeId.LINUX), - "macos": (OSTypeStr.MACOS, OSTypeId.MACOS), - "mac": (OSTypeStr.MACOS, OSTypeId.MACOS), - "android": (OSTypeStr.ANDROID, OSTypeId.ANDROID), - "ios": (OSTypeStr.IOS, OSTypeId.IOS), - } - def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self.context = PersistentJSON("device_context.json", self._data_path) @@ -85,64 +72,20 @@ def metadata(self) -> Metadata: @staticmethod def _to_epoch(value: str | None) -> float | None: """Convert an ISO 8601 timestamp to a Unix epoch float.""" - if not value: - return None - return isoparse(value).timestamp() + return mappers.to_epoch(value) @staticmethod def build_device_type(os_is_server: bool | None) -> tuple[DeviceTypeStr, DeviceTypeId]: """Map ``os_is_server`` to an OCSF device type.""" - if os_is_server: - return DeviceTypeStr.SERVER, DeviceTypeId.SERVER - - if os_is_server is False: - return DeviceTypeStr.DESKTOP, DeviceTypeId.DESKTOP - - return DeviceTypeStr.UNKNOWN, DeviceTypeId.UNKNOWN + return mappers.map_device_type(os_is_server) def build_operating_system(self, device: HolmDevice) -> OperatingSystem | None: """Map the Holm OS fields to an OCSF ``OperatingSystem`` object.""" - if device.os_name is None and device.os_family is None: - return None - - os_type: OSTypeStr = OSTypeStr.UNKNOWN - os_type_id: OSTypeId = OSTypeId.UNKNOWN - if device.os_family: - os_type, os_type_id = self.OS_FAMILY_MAP.get( - device.os_family.strip().lower(), (OSTypeStr.OTHER, OSTypeId.OTHER) - ) - - return OperatingSystem(name=device.os_name, type=os_type, type_id=os_type_id) + return mappers.build_operating_system(device.os_name, device.os_family) def build_network_interfaces(self, device: HolmDevice) -> list[NetworkInterface] | None: """Build the primary IPv4 and secondary IPv6 network interfaces.""" - network = device.network - if network is None: - return None - - interfaces: list[NetworkInterface] = [] - - if network.ip_address: - interfaces.append( - NetworkInterface( - ip=network.ip_address, - mac=network.mac_address, - hostname=device.hostname, - type=NetworkInterfaceTypeStr.WIRED, - type_id=NetworkInterfaceTypeId.WIRED, - ) - ) - - if network.ip_address_v6: - interfaces.append( - NetworkInterface( - ip=network.ip_address_v6, - type=NetworkInterfaceTypeStr.WIRED, - type_id=NetworkInterfaceTypeId.WIRED, - ) - ) - - return interfaces or None + return mappers.build_network_interfaces(device.network, device.hostname) def build_device(self, device: HolmDevice) -> Device: """Map a Holm device record to an OCSF ``Device`` object.""" diff --git a/HolmSecurity/holm_security/asset_connector/mappers.py b/HolmSecurity/holm_security/asset_connector/mappers.py new file mode 100644 index 000000000..c532edcdf --- /dev/null +++ b/HolmSecurity/holm_security/asset_connector/mappers.py @@ -0,0 +1,88 @@ +"""Shared helpers to map Holm Security device fields to OCSF objects. + +Used by both the device inventory connector and the vulnerability connector so +the device sub-object is built consistently from a Holm device record. +""" + +from dateutil.parser import isoparse +from sekoia_automation.asset_connector.models.ocsf.device import ( + DeviceTypeId, + DeviceTypeStr, + NetworkInterface, + NetworkInterfaceTypeId, + NetworkInterfaceTypeStr, + OperatingSystem, + OSTypeId, + OSTypeStr, +) + +from holm_security.asset_connector.models import HolmNetwork + +# Holm os_family -> (OCSF OSTypeStr, OCSF OSTypeId) +OS_FAMILY_MAP: dict[str, tuple[OSTypeStr, OSTypeId]] = { + "windows": (OSTypeStr.WINDOWS, OSTypeId.WINDOWS), + "linux": (OSTypeStr.LINUX, OSTypeId.LINUX), + "macos": (OSTypeStr.MACOS, OSTypeId.MACOS), + "mac": (OSTypeStr.MACOS, OSTypeId.MACOS), + "android": (OSTypeStr.ANDROID, OSTypeId.ANDROID), + "ios": (OSTypeStr.IOS, OSTypeId.IOS), +} + + +def to_epoch(value: str | None) -> float | None: + """Convert an ISO 8601 timestamp to a Unix epoch float.""" + if not value: + return None + return isoparse(value).timestamp() + + +def map_device_type(os_is_server: bool | None) -> tuple[DeviceTypeStr, DeviceTypeId]: + """Map ``os_is_server`` to an OCSF device type.""" + if os_is_server: + return DeviceTypeStr.SERVER, DeviceTypeId.SERVER + if os_is_server is False: + return DeviceTypeStr.DESKTOP, DeviceTypeId.DESKTOP + return DeviceTypeStr.UNKNOWN, DeviceTypeId.UNKNOWN + + +def build_operating_system(os_name: str | None, os_family: str | None) -> OperatingSystem | None: + """Map the Holm OS fields to an OCSF ``OperatingSystem`` object.""" + if os_name is None and os_family is None: + return None + + os_type: OSTypeStr = OSTypeStr.UNKNOWN + os_type_id: OSTypeId = OSTypeId.UNKNOWN + if os_family: + os_type, os_type_id = OS_FAMILY_MAP.get(os_family.strip().lower(), (OSTypeStr.OTHER, OSTypeId.OTHER)) + + return OperatingSystem(name=os_name, type=os_type, type_id=os_type_id) + + +def build_network_interfaces(network: HolmNetwork | None, hostname: str | None) -> list[NetworkInterface] | None: + """Build the primary IPv4 and secondary IPv6 network interfaces from a Holm network block.""" + if network is None: + return None + + interfaces: list[NetworkInterface] = [] + + if network.ip_address: + interfaces.append( + NetworkInterface( + ip=network.ip_address, + mac=network.mac_address, + hostname=hostname, + type=NetworkInterfaceTypeStr.WIRED, + type_id=NetworkInterfaceTypeId.WIRED, + ) + ) + + if network.ip_address_v6: + interfaces.append( + NetworkInterface( + ip=network.ip_address_v6, + type=NetworkInterfaceTypeStr.WIRED, + type_id=NetworkInterfaceTypeId.WIRED, + ) + ) + + return interfaces or None diff --git a/HolmSecurity/holm_security/asset_connector/vulnerability_assets.py b/HolmSecurity/holm_security/asset_connector/vulnerability_assets.py new file mode 100644 index 000000000..38586abf9 --- /dev/null +++ b/HolmSecurity/holm_security/asset_connector/vulnerability_assets.py @@ -0,0 +1,361 @@ +from collections.abc import Generator +from datetime import datetime +from functools import cached_property + +from dateutil.parser import isoparse +from pydantic import BaseModel, ValidationError +from requests.exceptions import RequestException +from sekoia_automation.asset_connector import AssetConnector +from sekoia_automation.asset_connector.models.ocsf.base import Metadata, Product +from sekoia_automation.asset_connector.models.ocsf.device import Device, DeviceTypeId, DeviceTypeStr +from sekoia_automation.asset_connector.models.ocsf.vulnerability import ( + CVE, + CVSS, + FindingInformation, + VulnerabilityDetails, + VulnerabilityOCSFModel, +) +from sekoia_automation.storage import PersistentJSON + +from holm_security.asset_connector import mappers +from holm_security.asset_connector.models import HolmDevice, HolmDevicePage +from holm_security.asset_connector.vulnerability_models import ( + HolmVulnDefinition, + HolmVulnerabilityFinding, + HolmVulnerabilityPage, + HolmVulnHost, +) +from holm_security.client import ApiClient + + +class HolmSecurityVulnerabilityAssetConnector(AssetConnector): + """Collect Holm Security vulnerability findings, correlate them with managed + devices, and map them to the OCSF Vulnerability Finding model.""" + + # API configuration + DEVICES_ENDPOINT: str = "/v2/devices" + VULNERABILITIES_ENDPOINT: str = "/v2/net-assets/report/vulnerabilities/" + DEFAULT_PAGE_SIZE: int = 100 + REQUEST_TIMEOUT: int = 60 + + # Product / metadata constants + PRODUCT_NAME: str = "Holm Security" + METADATA_VERSION: str = "1.6.0" + + # OCSF Vulnerability Finding constants + CATEGORY_NAME: str = "Findings" + CATEGORY_UID: int = 2 + CLASS_NAME: str = "Vulnerability Finding" + CLASS_UID: int = 2002 + + # Holm severity int -> (OCSF severity string, OCSF severity_id) + SEVERITY_MAP: dict[int, tuple[str, int]] = { + 0: ("Informational", 1), + 1: ("Low", 2), + 2: ("Medium", 3), + 3: ("High", 4), + 4: ("Critical", 5), + } + + # Holm status int -> (activity_id, activity_name, type_uid) + STATUS_MAP: dict[int, tuple[int, str, int]] = { + 1: (1, "Create", 200201), + 2: (3, "Close", 200203), + } + DEFAULT_STATUS: tuple[int, str, int] = (1, "Create", 200201) + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.context = PersistentJSON("vulnerability_context.json", self._data_path) + # Cache of managed devices, keyed by lowercased hostname / device name / IP. + self._agents_cache: dict[str, HolmDevice] = {} + + @property + def most_recent_last_detected(self) -> str | None: + with self.context as cache: + return cache.get("most_recent_last_detected") + + @cached_property + def base_url(self) -> str: + return str(self.module.configuration["base_url"]).rstrip("/") + + @cached_property + def client(self) -> ApiClient: + return ApiClient(base_url=self.base_url, token=self.module.configuration["api_token"]) + + @cached_property + def metadata(self) -> Metadata: + return Metadata( + product=Product(name=self.PRODUCT_NAME, vendor_name=self.PRODUCT_NAME), + version=self.METADATA_VERSION, + ) + + def map_severity(self, severity: int | None) -> tuple[str, int]: + """Map the Holm integer severity to an OCSF severity string and id.""" + if severity is None: + return "Unknown", 0 + return self.SEVERITY_MAP.get(severity, ("Unknown", 0)) + + def map_status(self, status: int | None) -> tuple[int, str, int]: + """Map the Holm integer status to (activity_id, activity_name, type_uid).""" + if status is None: + return self.DEFAULT_STATUS + return self.STATUS_MAP.get(status, self.DEFAULT_STATUS) + + # --- pagination ----------------------------------------------------------- + + def _fetch_pages(self, endpoint: str, model_cls: type[BaseModel]) -> Generator[BaseModel, None, None]: + """Fetch pages from a Holm endpoint, following ``next`` until it is null.""" + url: str | None = f"{self.base_url}{endpoint}" + params: dict[str, int] | None = {"limit": self.DEFAULT_PAGE_SIZE} + + try: + while url and self.running: + response = self.client.get(url, params=params, timeout=self.REQUEST_TIMEOUT) + response.raise_for_status() + + page = model_cls.model_validate(response.json()) + yield page + + url = getattr(page, "next", None) + # The `next` URL already carries the pagination query parameters. + params = None + except RequestException as error: + self.log(message=f"Holm Security API request failed for {endpoint}: {error}", level="error") + raise + + # --- step 1: device inventory cache -------------------------------------- + + def build_agents_cache(self) -> dict[str, HolmDevice]: + """Build a lookup of managed devices keyed by hostname / device name / IP.""" + cache: dict[str, HolmDevice] = {} + device_count = 0 + + for page in self._fetch_pages(self.DEVICES_ENDPOINT, HolmDevicePage): + assert isinstance(page, HolmDevicePage) + for device in page.results: + device_count += 1 + if device.hostname: + cache[device.hostname.lower()] = device + if device.device_name: + cache[device.device_name.lower()] = device + if device.network and device.network.ip_address: + cache[device.network.ip_address] = device + + self.log(message=f"Built device cache with {device_count} devices", level="info") + return cache + + def lookup_device(self, host: HolmVulnHost | None) -> HolmDevice | None: + """Correlate a finding host to a cached managed device.""" + if host is None: + return None + if host.name: + match = self._agents_cache.get(host.name.lower()) + if match is not None: + return match + if host.ip: + return self._agents_cache.get(host.ip) + return None + + # --- device sub-object ---------------------------------------------------- + + def build_device_from_cache(self, cached: HolmDevice, host: HolmVulnHost | None) -> Device: + """Build a full OCSF Device from a cached managed device.""" + device_type, device_type_id = mappers.map_device_type(cached.os_is_server) + network = cached.network + host_name = host.name if host else None + host_ip = host.ip if host else None + + return Device( + uid=cached.uid, + type=device_type, + type_id=device_type_id, + hostname=cached.hostname or host_name or "", + name=cached.device_name or host_name, + ip=(network.ip_address if network else None) or host_ip, + os=mappers.build_operating_system(cached.os_name, cached.os_family), + network_interfaces=mappers.build_network_interfaces(network, cached.hostname), + first_seen_time=mappers.to_epoch(cached.created), + last_seen_time=mappers.to_epoch(cached.last_sync), + is_managed=True, + vendor_name=self.PRODUCT_NAME, + ) + + def build_minimal_device(self, host: HolmVulnHost | None) -> Device: + """Build a minimal OCSF Device from the finding host when no agent matches.""" + host_name = host.name if host else None + host_ip = host.ip if host else None + + return Device( + uid=host_name or host_ip or "unknown", + type=DeviceTypeStr.UNKNOWN, + type_id=DeviceTypeId.UNKNOWN, + hostname=host_name or "", + name=host_name, + ip=host_ip, + is_managed=False, + vendor_name=self.PRODUCT_NAME, + ) + + def build_device(self, host: HolmVulnHost | None) -> Device: + cached = self.lookup_device(host) + if cached is not None: + return self.build_device_from_cache(cached, host) + return self.build_minimal_device(host) + + # --- vulnerability sub-objects ------------------------------------------- + + @staticmethod + def _build_cvss(definition: HolmVulnDefinition | None) -> list[CVSS] | None: + inner = definition.vuln if definition else None + if inner is None: + return None + + cvss: list[CVSS] = [] + if inner.base_score_v2: + cvss.append(CVSS(version="2.0", base_score=float(inner.base_score_v2))) + if inner.base_score_v3: + cvss.append(CVSS(version="3.0", base_score=float(inner.base_score_v3))) + return cvss or None + + def build_vulnerabilities( + self, definition: HolmVulnDefinition | None, severity: str + ) -> list[VulnerabilityDetails]: + """Build the OCSF vulnerabilities list, one entry per CVE (or one entry with no CVE).""" + inner = definition.vuln if definition else None + title = definition.name if definition else None + summary = inner.summary if inner else None + is_fix_available = inner.patch_avail if inner else None + cvss = self._build_cvss(definition) + references = [ref for ref in [inner.vendor_ref, inner.exploit_ref] if ref] if inner else None + cves = [cve for cve in (definition.cves or []) if cve] if definition else [] + + details: list[VulnerabilityDetails] = [] + for cve_id in cves: + details.append( + VulnerabilityDetails( + cve=CVE(uid=cve_id, type="CVE", cvss=cvss, desc=summary, title=title), + title=title, + desc=summary, + references=references or None, + severity=severity, + vendor_name=self.PRODUCT_NAME, + is_fix_available=is_fix_available, + ) + ) + + if not details: + details.append( + VulnerabilityDetails( + title=title, + desc=summary, + references=references or None, + severity=severity, + vendor_name=self.PRODUCT_NAME, + is_fix_available=is_fix_available, + ) + ) + + return details + + def build_finding_info(self, finding: HolmVulnerabilityFinding) -> FindingInformation: + definition = finding.vuln + inner = definition.vuln if definition else None + category = definition.vuln_category.name if definition and definition.vuln_category else None + + def to_int_epoch(value: str | None) -> int | None: + epoch = mappers.to_epoch(value) + return int(epoch) if epoch is not None else None + + return FindingInformation( + uid=(definition.hid if definition and definition.hid else "unknown"), + title=definition.name if definition else None, + desc=inner.summary if inner else None, + types=[category] if category else None, + created_time=to_int_epoch(definition.created if definition else None), + first_seen_time=to_int_epoch(finding.created_timestamp), + last_seen_time=to_int_epoch(finding.last_detected), + ) + + def map_fields(self, finding: HolmVulnerabilityFinding) -> VulnerabilityOCSFModel: + """Map a Holm vulnerability finding to the OCSF Vulnerability Finding model.""" + event_time = mappers.to_epoch(finding.last_detected) + if event_time is None: + raise ValueError("Vulnerability finding has no last_detected timestamp") + + severity, severity_id = self.map_severity(finding.severity) + activity_id, activity_name, type_uid = self.map_status(finding.status) + + return VulnerabilityOCSFModel( + activity_id=activity_id, + activity_name=activity_name, + category_name=self.CATEGORY_NAME, + category_uid=self.CATEGORY_UID, + class_name=self.CLASS_NAME, + class_uid=self.CLASS_UID, + type_name=f"{self.CLASS_NAME}: {activity_name}", + type_uid=type_uid, + severity=severity, + severity_id=severity_id, + time=event_time, + metadata=self.metadata, + finding_info=self.build_finding_info(finding), + vulnerabilities=self.build_vulnerabilities(finding.vuln, severity), + device=self.build_device(finding.host), + ) + + # --- driver --------------------------------------------------------------- + + def get_assets(self) -> Generator[VulnerabilityOCSFModel, None, None]: + # Step 1: load the managed device inventory into a cache. + self._agents_cache = self.build_agents_cache() + + checkpoint = self.most_recent_last_detected + checkpoint_dt: datetime | None = isoparse(checkpoint) if checkpoint else None + max_dt: datetime | None = checkpoint_dt + max_raw: str | None = checkpoint + + generated = 0 + skipped = 0 + + # Steps 2 & 3: fetch vulnerability findings, correlate, and emit. + for page in self._fetch_pages(self.VULNERABILITIES_ENDPOINT, HolmVulnerabilityPage): + assert isinstance(page, HolmVulnerabilityPage) + for finding in page.results: + finding_dt = isoparse(finding.last_detected) if finding.last_detected else None + + # Client-side checkpoint filter: skip findings not detected since last run. + if checkpoint_dt is not None and finding_dt is not None and finding_dt <= checkpoint_dt: + skipped += 1 + continue + + if finding_dt is not None and (max_dt is None or finding_dt > max_dt): + max_dt = finding_dt + max_raw = finding.last_detected + + try: + asset = self.map_fields(finding) + except (ValueError, ValidationError) as error: + skipped += 1 + self.log(message=f"Skipping vulnerability finding: {error}", level="warning") + continue + + generated += 1 + yield asset + + # Persist the new checkpoint only after the full run has been consumed. + if max_raw and max_raw != checkpoint: + self._latest_time = max_raw + + self.log( + message=f"Vulnerability generation complete - generated: {generated}, skipped: {skipped}", + level="info", + ) + + def update_checkpoint(self) -> None: + if self._latest_time: + with self.context as cache: + cache["most_recent_last_detected"] = self._latest_time + self.log(message=f"Checkpoint updated to {self._latest_time}", level="debug") + else: + self.log(message="No checkpoint update needed", level="debug") diff --git a/HolmSecurity/holm_security/asset_connector/vulnerability_mapping.yml b/HolmSecurity/holm_security/asset_connector/vulnerability_mapping.yml new file mode 100644 index 000000000..29eb0aba3 --- /dev/null +++ b/HolmSecurity/holm_security/asset_connector/vulnerability_mapping.yml @@ -0,0 +1,257 @@ +mapping: + ocsf_class: "Vulnerability Finding" + class_uid: 2002 + ocsf_version: "1.6.0" + + samples: + - name: "Holm Security Vulnerability Finding Sample" + description: "Single finding from GET /v2/net-assets/report/vulnerabilities/." + data: | + { + "asset": "00000000-0000-0000-0000-000000000000", + "created_timestamp": "2026-07-01T14:15:22Z", + "last_detected": "2026-07-10T14:15:22Z", + "modified_timestamp": "2026-07-10T14:15:22Z", + "severity": 4, + "status": 1, + "host": { + "name": "desktop-example01", + "ip": "192.0.2.10" + }, + "vuln": { + "hid": "hid-0001", + "name": "Example Remote Code Execution", + "created": "2026-06-01T00:00:00Z", + "modified": "2026-06-15T00:00:00Z", + "score": 9, + "cves": ["CVE-2026-0001"], + "vuln": { + "base_score_v2": 9.0, + "base_score_v3": 9.8, + "patch_avail": true, + "summary": "A remote code execution vulnerability in the example service.", + "solution": "Apply the vendor patch.", + "vendor_ref": "https://example.com/advisory/0001" + }, + "vuln_category": { + "id": 1, + "name": "Remote Code Execution" + } + } + } + + fields: + # OCSF EVENT CLASSIFICATION + - source: "status" + target: "activity_id" + type: "integer" + logic: "1 (Open) -> 1 Create; 2 (Fixed) -> 3 Close; other -> 1 Create" + description: "OCSF activity ID" + + - source: "status" + target: "activity_name" + type: "string" + logic: "1 -> 'Create'; 2 -> 'Close'; other -> 'Create'" + description: "OCSF activity name" + + - source: "static: Findings" + target: "category_name" + type: "string" + logic: "Always 'Findings'" + description: "OCSF category name" + + - source: "static: 2" + target: "category_uid" + type: "integer" + logic: "Always 2 for Findings category" + description: "OCSF category UID" + + - source: "static: Vulnerability Finding" + target: "class_name" + type: "string" + logic: "Always 'Vulnerability Finding'" + description: "OCSF class name" + + - source: "static: 2002" + target: "class_uid" + type: "integer" + logic: "Always 2002 for Vulnerability Finding" + description: "OCSF class UID" + + - source: "status" + target: "type_uid" + type: "integer" + logic: "1 -> 200201; 2 -> 200203; other -> 200201" + description: "OCSF type UID" + + - source: "status" + target: "type_name" + type: "string" + logic: "'Vulnerability Finding: ' + activity_name" + description: "OCSF type name" + + # SEVERITY + - source: "severity" + target: "severity" + type: "string" + logic: "0->Informational, 1->Low, 2->Medium, 3->High, 4->Critical, else Unknown" + description: "OCSF severity string" + + - source: "severity" + target: "severity_id" + type: "integer" + logic: "severity + 1 if in [0,4], else 0" + description: "OCSF severity id" + + # TIMESTAMP + - source: "last_detected" + target: "time" + type: "timestamp" + logic: "Convert ISO 8601 last_detected to Unix epoch" + description: "OCSF event timestamp" + + # METADATA + - source: "static: Holm Security" + target: "metadata.product.name" + type: "string" + logic: "Always 'Holm Security'" + description: "Source product name" + + - source: "static: Holm Security" + target: "metadata.product.vendor_name" + type: "string" + logic: "Always 'Holm Security'" + description: "Vendor name" + + - source: "static: 1.6.0" + target: "metadata.version" + type: "string" + logic: "Fixed OCSF schema version" + description: "OCSF schema version" + + # FINDING INFORMATION + - source: "vuln.hid" + target: "finding_info.uid" + type: "string" + logic: "Vulnerability definition id; 'unknown' fallback" + description: "Finding unique identifier" + + - source: "vuln.name" + target: "finding_info.title" + type: "string" + logic: "Direct mapping of vulnerability name" + description: "Finding title" + + - source: "vuln.vuln.summary" + target: "finding_info.desc" + type: "string" + logic: "Direct mapping of vulnerability summary" + description: "Finding description" + + # VULNERABILITY DETAILS + - source: "vuln.cves[]" + target: "vulnerabilities[].cve.uid" + type: "string" + logic: "One VulnerabilityDetails entry per CVE" + description: "CVE identifier" + + - source: "vuln.vuln.base_score_v2" + target: "vulnerabilities[].cve.cvss[].base_score" + type: "float" + logic: "CVSS v2.0 base score" + description: "CVSS v2 base score" + + - source: "vuln.vuln.base_score_v3" + target: "vulnerabilities[].cve.cvss[].base_score" + type: "float" + logic: "CVSS v3.0 base score" + description: "CVSS v3 base score" + + - source: "vuln.vuln.patch_avail" + target: "vulnerabilities[].is_fix_available" + type: "boolean" + logic: "Direct mapping of patch availability" + description: "Whether a fix is available" + + # DEVICE (correlated from device inventory cache) + - source: "agents_cache[host].uid" + target: "device.uid" + type: "string" + logic: "Cached device uid; minimal device uses host.name/host.ip fallback" + description: "Device unique identifier" + + - source: "agents_cache[host].os_is_server" + target: "device.type / device.type_id" + type: "string" + logic: "true -> Server, false -> Desktop, null/no-match -> Unknown" + description: "Device type" + + - source: "host.name / host.ip" + target: "device.hostname / device.ip" + type: "string" + logic: "Prefer cached values; fallback to finding host fields" + description: "Device hostname and IP" + + - source: "cache lookup" + target: "device.is_managed" + type: "boolean" + logic: "True when a matching agent is found in the cache, else False" + description: "Whether the device is agent-managed" + + ocsf_samples: + - name: "Vulnerability Finding: Create" + description: "Transformed Holm Security finding to OCSF Vulnerability Finding event" + data: | + { + "activity_id": 1, + "activity_name": "Create", + "category_name": "Findings", + "category_uid": 2, + "class_name": "Vulnerability Finding", + "class_uid": 2002, + "type_name": "Vulnerability Finding: Create", + "type_uid": 200201, + "severity": "Critical", + "severity_id": 5, + "time": 1752156922.0, + "metadata": { + "product": { + "name": "Holm Security", + "vendor_name": "Holm Security" + }, + "version": "1.6.0" + }, + "finding_info": { + "uid": "hid-0001", + "title": "Example Remote Code Execution", + "desc": "A remote code execution vulnerability in the example service." + }, + "vulnerabilities": [ + { + "title": "Example Remote Code Execution", + "desc": "A remote code execution vulnerability in the example service.", + "severity": "Critical", + "vendor_name": "Holm Security", + "is_fix_available": true, + "references": ["https://example.com/advisory/0001"], + "cve": { + "uid": "CVE-2026-0001", + "type": "CVE", + "cvss": [ + {"version": "2.0", "base_score": 9.0}, + {"version": "3.0", "base_score": 9.8} + ] + } + } + ], + "device": { + "uid": "0123456789abcdef0123456789abcdef", + "name": "DESKTOP-EXAMPLE01", + "hostname": "desktop-example01", + "type": "Desktop", + "type_id": 2, + "ip": "192.0.2.10", + "is_managed": true, + "vendor_name": "Holm Security" + } + } diff --git a/HolmSecurity/holm_security/asset_connector/vulnerability_models.py b/HolmSecurity/holm_security/asset_connector/vulnerability_models.py new file mode 100644 index 000000000..578e48402 --- /dev/null +++ b/HolmSecurity/holm_security/asset_connector/vulnerability_models.py @@ -0,0 +1,77 @@ +from pydantic import BaseModel, ConfigDict + + +class HolmVulnHost(BaseModel): + """Host context attached to a vulnerability finding.""" + + model_config = ConfigDict(extra="allow") + + name: str | None = None + ip: str | None = None + + +class HolmVulnInner(BaseModel): + """CVSS / detection details nested under ``vuln.vuln``.""" + + model_config = ConfigDict(extra="allow") + + base_score_v2: float | None = None + base_score_v3: float | None = None + patch_avail: bool | None = None + summary: str | None = None + solution: str | None = None + insight: str | None = None + impact: str | None = None + vendor_ref: str | None = None + exploit_ref: str | None = None + software: str | None = None + + +class HolmVulnCategory(BaseModel): + model_config = ConfigDict(extra="allow") + + id: int | None = None + name: str | None = None + + +class HolmVulnDefinition(BaseModel): + """Vulnerability definition nested under the finding's ``vuln`` field.""" + + model_config = ConfigDict(extra="allow") + + hid: str | None = None + name: str | None = None + created: str | None = None + modified: str | None = None + score: float | None = None + cves: list[str] | None = None + vuln: HolmVulnInner | None = None + vuln_category: HolmVulnCategory | None = None + + +class HolmVulnerabilityFinding(BaseModel): + """A single vulnerability finding returned by the report endpoint.""" + + model_config = ConfigDict(extra="allow") + + asset: str | None = None + created_timestamp: str | None = None + last_detected: str | None = None + modified_timestamp: str | None = None + severity: int | None = None + status: int | None = None + solution: str | None = None + details: str | None = None + host: HolmVulnHost | None = None + vuln: HolmVulnDefinition | None = None + + +class HolmVulnerabilityPage(BaseModel): + """Paginated response envelope for the vulnerabilities report endpoint.""" + + model_config = ConfigDict(extra="allow") + + count: int = 0 + next: str | None = None + previous: str | None = None + results: list[HolmVulnerabilityFinding] = [] diff --git a/HolmSecurity/main.py b/HolmSecurity/main.py index bf25248ef..e448fd985 100644 --- a/HolmSecurity/main.py +++ b/HolmSecurity/main.py @@ -2,9 +2,11 @@ from holm_security.account_validator import HolmSecurityAccountValidator from holm_security.asset_connector.device_assets import HolmSecurityDeviceAssetConnector +from holm_security.asset_connector.vulnerability_assets import HolmSecurityVulnerabilityAssetConnector if __name__ == "__main__": module = Module() module.register_account_validator(HolmSecurityAccountValidator) module.register(HolmSecurityDeviceAssetConnector, "holm_security_device_asset_connector") + module.register(HolmSecurityVulnerabilityAssetConnector, "holm_security_vulnerability_asset_connector") module.run() diff --git a/HolmSecurity/manifest.json b/HolmSecurity/manifest.json index cc173f024..553e6c3de 100644 --- a/HolmSecurity/manifest.json +++ b/HolmSecurity/manifest.json @@ -27,7 +27,7 @@ "name": "Holm Security", "uuid": "23886dfe-89a9-4522-b1fb-6e2837e15390", "slug": "holm-security", - "version": "1.0.0", + "version": "1.1.0", "categories": [ "Network" ], diff --git a/HolmSecurity/tests/asset_connector/test_vulnerability_assets.py b/HolmSecurity/tests/asset_connector/test_vulnerability_assets.py new file mode 100644 index 000000000..4b017286d --- /dev/null +++ b/HolmSecurity/tests/asset_connector/test_vulnerability_assets.py @@ -0,0 +1,238 @@ +from unittest.mock import Mock + +import pytest +from sekoia_automation.asset_connector.models.ocsf.vulnerability import VulnerabilityOCSFModel +from sekoia_automation.module import Module + +from holm_security.asset_connector.models import HolmDevice +from holm_security.asset_connector.vulnerability_assets import HolmSecurityVulnerabilityAssetConnector +from holm_security.asset_connector.vulnerability_models import HolmVulnerabilityFinding, HolmVulnHost + +BASE_URL = "https://se-api.holmsecurity.com" +DEVICES_URL = f"{BASE_URL}/v2/devices" +VULNS_URL = f"{BASE_URL}/v2/net-assets/report/vulnerabilities/" + + +def make_device(**overrides) -> dict: + device = { + "uid": "0123456789abcdef0123456789abcdef", + "device_name": "DESKTOP-EXAMPLE01", + "hostname": "desktop-example01", + "last_sync": "2026-07-01T20:30:36Z", + "created": "2026-06-01T00:00:00Z", + "os_is_server": False, + "os_family": "windows", + "os_name": "Microsoft Windows Server 2025 Datacenter", + "network": {"ip_address": "192.0.2.10", "ip_address_v6": None, "mac_address": "00:00:5E:00:53:00"}, + } + device.update(overrides) + return device + + +def make_finding(**overrides) -> dict: + finding = { + "asset": "00000000-0000-0000-0000-000000000000", + "created_timestamp": "2026-07-01T14:15:22Z", + "last_detected": "2026-07-10T14:15:22Z", + "modified_timestamp": "2026-07-10T14:15:22Z", + "severity": 4, + "status": 1, + "host": {"name": "desktop-example01", "ip": "192.0.2.10"}, + "vuln": { + "hid": "hid-0001", + "name": "Example Remote Code Execution", + "created": "2026-06-01T00:00:00Z", + "cves": ["CVE-2026-0001"], + "vuln": { + "base_score_v2": 9.0, + "base_score_v3": 9.8, + "patch_avail": True, + "summary": "A remote code execution vulnerability.", + "vendor_ref": "https://example.com/advisory/0001", + }, + "vuln_category": {"id": 1, "name": "Remote Code Execution"}, + }, + } + finding.update(overrides) + return finding + + +@pytest.fixture +def connector(symphony_storage): + module = Module() + module.configuration = {"base_url": BASE_URL, "api_token": "fake_api_token"} + + connector = HolmSecurityVulnerabilityAssetConnector(module=module, data_path=symphony_storage) + connector.configuration = { + "sekoia_base_url": "https://sekoia.io", + "sekoia_api_key": "fake_sekoia_api_key", + "frequency": 60, + } + connector.log = Mock() + connector.log_exception = Mock() + yield connector + + +def test_map_severity(connector): + assert connector.map_severity(0) == ("Informational", 1) + assert connector.map_severity(4) == ("Critical", 5) + assert connector.map_severity(9) == ("Unknown", 0) + assert connector.map_severity(None) == ("Unknown", 0) + + +def test_map_status(connector): + assert connector.map_status(1) == (1, "Create", 200201) + assert connector.map_status(2) == (3, "Close", 200203) + assert connector.map_status(7) == (1, "Create", 200201) + assert connector.map_status(None) == (1, "Create", 200201) + + +def test_map_fields_full(connector): + connector._agents_cache = {"desktop-example01": HolmDevice.model_validate(make_device())} + + asset = connector.map_fields(HolmVulnerabilityFinding.model_validate(make_finding())) + + assert isinstance(asset, VulnerabilityOCSFModel) + assert asset.class_uid == 2002 + assert asset.category_uid == 2 + assert asset.activity_id == 1 + assert asset.type_uid == 200201 + assert asset.type_name == "Vulnerability Finding: Create" + assert asset.severity == "Critical" + assert asset.severity_id == 5 + assert asset.metadata.product.vendor_name == "Holm Security" + assert asset.metadata.version == "1.6.0" + + assert asset.finding_info.uid == "hid-0001" + assert asset.finding_info.title == "Example Remote Code Execution" + + assert len(asset.vulnerabilities) == 1 + vuln = asset.vulnerabilities[0] + assert vuln.cve.uid == "CVE-2026-0001" + assert vuln.is_fix_available is True + assert {c.version for c in vuln.cve.cvss} == {"2.0", "3.0"} + + # Correlated to the managed device from cache. + assert asset.device.uid == "0123456789abcdef0123456789abcdef" + assert asset.device.is_managed is True + assert asset.device.type_id == 2 # Desktop + + +def test_map_fields_closed_status(connector): + asset = connector.map_fields(HolmVulnerabilityFinding.model_validate(make_finding(status=2))) + assert asset.activity_id == 3 + assert asset.type_uid == 200203 + assert asset.type_name == "Vulnerability Finding: Close" + + +def test_build_vulnerabilities_without_cve(connector): + finding = make_finding() + finding["vuln"]["cves"] = [] + details = connector.build_vulnerabilities(HolmVulnerabilityFinding.model_validate(finding).vuln, "Critical") + assert len(details) == 1 + assert details[0].cve is None + assert details[0].severity == "Critical" + + +def test_lookup_by_ip(connector): + connector._agents_cache = {"192.0.2.10": HolmDevice.model_validate(make_device())} + host = HolmVulnHost(name="unknown-host", ip="192.0.2.10") + assert connector.lookup_device(host) is not None + + +def test_minimal_device_when_no_match(connector): + connector._agents_cache = {} + asset = connector.map_fields(HolmVulnerabilityFinding.model_validate(make_finding())) + assert asset.device.is_managed is False + assert asset.device.hostname == "desktop-example01" + assert asset.device.ip == "192.0.2.10" + assert asset.device.type_id == 0 # Unknown + + +def test_map_fields_without_last_detected_raises(connector): + with pytest.raises(ValueError): + connector.map_fields(HolmVulnerabilityFinding.model_validate(make_finding(last_detected=None))) + + +def test_get_assets_end_to_end(connector, requests_mock): + requests_mock.get( + DEVICES_URL, + json={"count": 1, "next": None, "previous": None, "results": [make_device()]}, + ) + managed = make_finding() + unmanaged = make_finding(host={"name": "other-host", "ip": "203.0.113.9"}) + requests_mock.get( + VULNS_URL, + json={"count": 2, "next": None, "previous": None, "results": [managed, unmanaged]}, + ) + + assets = list(connector.get_assets()) + + assert len(assets) == 2 + assert assets[0].device.is_managed is True + assert assets[1].device.is_managed is False + assert connector._latest_time == "2026-07-10T14:15:22Z" + + +def test_get_assets_checkpoint_filter(connector, requests_mock): + with connector.context as cache: + cache["most_recent_last_detected"] = "2026-07-05T00:00:00Z" + + requests_mock.get(DEVICES_URL, json={"count": 0, "next": None, "previous": None, "results": []}) + old = make_finding(last_detected="2026-07-01T00:00:00Z") + new = make_finding(last_detected="2026-07-10T00:00:00Z") + requests_mock.get( + VULNS_URL, + json={"count": 2, "next": None, "previous": None, "results": [old, new]}, + ) + + assets = list(connector.get_assets()) + + assert len(assets) == 1 + assert connector._latest_time == "2026-07-10T00:00:00Z" + + +def test_get_assets_paginates_vulns(connector, requests_mock): + requests_mock.get(DEVICES_URL, json={"count": 0, "next": None, "previous": None, "results": []}) + page1 = {"count": 2, "next": f"{VULNS_URL}?offset=1", "previous": None, "results": [make_finding()]} + page2 = {"count": 2, "next": None, "previous": None, "results": [make_finding()]} + requests_mock.get(VULNS_URL, json=page1) + requests_mock.get(f"{VULNS_URL}?offset=1", json=page2) + + assets = list(connector.get_assets()) + assert len(assets) == 2 + + +def test_update_checkpoint_persists(connector): + connector._latest_time = "2026-07-10T14:15:22Z" + connector.update_checkpoint() + assert connector.most_recent_last_detected == "2026-07-10T14:15:22Z" + + +def test_update_checkpoint_noop_when_unset(connector): + connector._latest_time = None + connector.update_checkpoint() + assert connector.most_recent_last_detected is None + + +def test_lookup_device_no_match(connector): + connector._agents_cache = {} + assert connector.lookup_device(None) is None + assert connector.lookup_device(HolmVulnHost(name="nope", ip="203.0.113.9")) is None + + +def test_get_assets_skips_finding_without_timestamp(connector, requests_mock): + requests_mock.get(DEVICES_URL, json={"count": 0, "next": None, "previous": None, "results": []}) + good = make_finding() + bad = make_finding(last_detected=None) + requests_mock.get(VULNS_URL, json={"count": 2, "next": None, "previous": None, "results": [good, bad]}) + + assets = list(connector.get_assets()) + assert len(assets) == 1 + + +def test_get_assets_raises_on_api_error(connector, requests_mock): + requests_mock.get(DEVICES_URL, json={"detail": "Server error"}, status_code=500) + + with pytest.raises(Exception): + list(connector.get_assets()) diff --git a/HolmSecurity/tests/test_account_validator.py b/HolmSecurity/tests/test_account_validator.py index 97f1c9381..21160f911 100644 --- a/HolmSecurity/tests/test_account_validator.py +++ b/HolmSecurity/tests/test_account_validator.py @@ -6,14 +6,16 @@ from holm_security.account_validator import HolmSecurityAccountValidator -VALIDATION_URL = "https://se-api.holmsecurity.com/v2/devices" +BASE_URL = "https://se-api.holmsecurity.com" +DEVICES_URL = f"{BASE_URL}/v2/devices" +VULNS_URL = f"{BASE_URL}/v2/net-assets/report/vulnerabilities/" @pytest.fixture def account_validator(): module = Module() module.configuration = { - "base_url": "https://se-api.holmsecurity.com", + "base_url": BASE_URL, "api_token": "fake_api_token", } validator = HolmSecurityAccountValidator(module=module) @@ -23,44 +25,53 @@ def account_validator(): def test_validate_success(account_validator, requests_mock): - matcher = requests_mock.get(VALIDATION_URL, json={"count": 1, "results": []}, status_code=200) + devices = requests_mock.get(DEVICES_URL, json={"count": 1, "results": []}, status_code=200) + vulns = requests_mock.get(VULNS_URL, json={"count": 1, "results": []}, status_code=200) assert account_validator.validate() is True - # Ping call must use page_size=1 - assert matcher.last_request.qs.get("page_size") == ["1"] + assert devices.last_request.qs.get("page_size") == ["1"] + assert vulns.last_request.qs.get("limit") == ["1"] -def test_validate_authentication_failure(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, json={"detail": "Invalid token."}, status_code=401) +def test_validate_devices_authentication_failure(account_validator, requests_mock): + requests_mock.get(DEVICES_URL, json={"detail": "Invalid token."}, status_code=401) + requests_mock.get(VULNS_URL, json={"count": 1, "results": []}, status_code=200) + + assert account_validator.validate() is False + + +def test_validate_vulnerabilities_failure(account_validator, requests_mock): + requests_mock.get(DEVICES_URL, json={"count": 1, "results": []}, status_code=200) + requests_mock.get(VULNS_URL, json={"detail": "Forbidden"}, status_code=403) assert account_validator.validate() is False def test_validate_server_error(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, json={"detail": "Internal error"}, status_code=500) + requests_mock.get(DEVICES_URL, json={"detail": "Internal error"}, status_code=500) assert account_validator.validate() is False def test_validate_timeout(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, exc=requests.Timeout) + requests_mock.get(DEVICES_URL, exc=requests.Timeout) assert account_validator.validate() is False def test_validate_connection_error(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, exc=requests.ConnectionError) + requests_mock.get(DEVICES_URL, exc=requests.ConnectionError) assert account_validator.validate() is False def test_validate_unexpected_error(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, exc=ValueError("boom")) + requests_mock.get(DEVICES_URL, exc=ValueError("boom")) assert account_validator.validate() is False def test_validate_non_json_error_body(account_validator, requests_mock): - requests_mock.get(VALIDATION_URL, text="Forbidden", status_code=403) + requests_mock.get(DEVICES_URL, text="Forbidden", status_code=403) assert account_validator.validate() is False