Feature: Holm Security Vulnerabilities

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

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"]
+  }
+}

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

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."""

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