Feature: Holm Security Devices#2914
Open
vg-svitla wants to merge 3 commits into
Open
Conversation
Contributor
Reviewer's GuideIntroduce a Holm Security integration module that authenticates via API token, validates credentials, and exposes a device asset connector that reads /v2/devices, maps Holm device records into OCSF Device Inventory Info events, and persists a checkpoint for incremental collection, with tests and project configuration. Sequence diagram for Holm Security device asset collectionsequenceDiagram
actor Operator
participant Module
participant HolmSecurityDeviceAssetConnector as DeviceConnector
participant PersistentJSON as CheckpointStore
participant ApiClient
participant HolmSecurityAPI as HolmSecurity_API
Operator->>Module: run()
Module->>DeviceConnector: get_assets()
DeviceConnector->>CheckpointStore: read most_recent_last_sync
CheckpointStore-->>DeviceConnector: checkpoint
loop fetch device pages
DeviceConnector->>ApiClient: get(base_url + /v2/devices, page_size)
ApiClient->>HolmSecurity_API: GET /v2/devices
HolmSecurity_API-->>ApiClient: 200 JSON page
ApiClient-->>DeviceConnector: Response
DeviceConnector->>DeviceConnector: HolmDevicePage.model_validate(json)
loop for each HolmDevice in page.results
DeviceConnector->>DeviceConnector: [last_sync <= checkpoint?]
alt last_sync <= checkpoint
DeviceConnector->>DeviceConnector: skip device
else last_sync > checkpoint or no checkpoint
DeviceConnector->>DeviceConnector: map_fields(device)
DeviceConnector-->>Module: yield DeviceOCSFModel
end
end
end
DeviceConnector->>CheckpointStore: write most_recent_last_sync
DeviceConnector-->>Module: Asset generation complete
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- Consider initializing
self._latest_timeinHolmSecurityDeviceAssetConnector.__init__to avoid potentialAttributeErrorwhenupdate_checkpointis called before any assets are generated. - In
HolmSecurityDeviceAssetConnector._to_epochand the date parsing inget_assets, you may want to handle invalid timestamp formats explicitly to avoid uncaught parsing errors if Holm Security returns unexpected date strings. - For the Pydantic models (
HolmDevicePage.resultsand any other list fields), preferdefault_factory=listover=[]to avoid shared mutable defaults.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Consider initializing `self._latest_time` in `HolmSecurityDeviceAssetConnector.__init__` to avoid potential `AttributeError` when `update_checkpoint` is called before any assets are generated.
- In `HolmSecurityDeviceAssetConnector._to_epoch` and the date parsing in `get_assets`, you may want to handle invalid timestamp formats explicitly to avoid uncaught parsing errors if Holm Security returns unexpected date strings.
- For the Pydantic models (`HolmDevicePage.results` and any other list fields), prefer `default_factory=list` over `=[]` to avoid shared mutable defaults.
## Individual Comments
### Comment 1
<location path="HolmSecurity/holm_security/asset_connector/device_assets.py" line_range="221-225" />
<code_context>
+ skipped = 0
+
+ for page in self._fetch_device_pages():
+ for device in page.results:
+ device_dt = isoparse(device.last_sync) if device.last_sync else None
+
+ # Client-side checkpoint filter: skip devices not modified since last run.
</code_context>
<issue_to_address>
**suggestion:** Protect the checkpoint filter from invalid last_sync values to avoid runtime errors.
`isoparse(device.last_sync)` runs outside the existing `try/except`, so a malformed `last_sync` will raise `ValueError` and crash before the error handling is reached. Please wrap the parse in its own `try/except` and treat invalid timestamps as `None`, e.g.:
```python
device_dt: datetime | None = None
if device.last_sync:
try:
device_dt = isoparse(device.last_sync)
except ValueError:
self.log(message=f"Invalid last_sync on device {device.uid}: {device.last_sync}", level="warning")
```
This keeps the checkpoint filter resilient to bad timestamp data.
```suggestion
for page in self._fetch_device_pages():
for device in page.results:
device_dt: datetime | None = None
if device.last_sync:
try:
device_dt = isoparse(device.last_sync)
except ValueError:
self.log(
message=f"Invalid last_sync on device {device.uid}: {device.last_sync}",
level="warning",
)
# Client-side checkpoint filter: skip devices not modified since last run.
```
</issue_to_address>
### Comment 2
<location path="HolmSecurity/tests/asset_connector/test_device_assets.py" line_range="104-107" />
<code_context>
+ assert device.os.type_id == OSTypeId.WINDOWS
+
+
+def test_operating_system_unknown_family(connector):
+ device = connector.build_device(HolmDevice.model_validate(make_device(os_family="plan9")))
+ assert device.os.type == OSTypeStr.OTHER
+ assert device.os.type_id == OSTypeId.OTHER
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for `build_operating_system` handling of mixed-case and whitespace `os_family` values.
Current tests only cover lowercase `os_family` values and an unknown family, but the implementation normalizes via `strip().lower()`. Please add cases like `os_family=" Windows "` and `"MAC"` to assert they resolve to the correct `OSTypeStr`/`OSTypeId`, so the normalization behavior is explicitly covered.
Suggested implementation:
```python
def test_operating_system_windows(connector):
device = connector.build_device(HolmDevice.model_validate(make_device()))
assert device.os is not None
assert device.os.name == "Microsoft Windows Server 2025 Datacenter"
assert device.os.type == OSTypeStr.WINDOWS
assert device.os.type_id == OSTypeId.WINDOWS
def test_operating_system_windows_mixed_case_whitespace(connector):
device = connector.build_device(
HolmDevice.model_validate(make_device(os_family=" Windows "))
)
assert device.os is not None
assert device.os.type == OSTypeStr.WINDOWS
assert device.os.type_id == OSTypeId.WINDOWS
def test_operating_system_mac_mixed_case(connector):
device = connector.build_device(
HolmDevice.model_validate(make_device(os_family="MAC"))
)
assert device.os is not None
assert device.os.type == OSTypeStr.MAC_OS
assert device.os.type_id == OSTypeId.MAC_OS
import pytest
```
1. Verify the exact enum members for macOS in `OSTypeStr` and `OSTypeId` (e.g. they might be `MAC_OS`, `MAC_OS_X`, or similar) and adjust the assertions in `test_operating_system_mac_mixed_case` accordingly.
2. If there is a dedicated factory or fixture for macOS devices (rather than using `make_device(os_family="MAC")`), update the test to use it to stay consistent with the rest of the test suite.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+221
to
+225
| for page in self._fetch_device_pages(): | ||
| for device in page.results: | ||
| device_dt = isoparse(device.last_sync) if device.last_sync else None | ||
|
|
||
| # Client-side checkpoint filter: skip devices not modified since last run. |
Contributor
There was a problem hiding this comment.
suggestion: Protect the checkpoint filter from invalid last_sync values to avoid runtime errors.
isoparse(device.last_sync) runs outside the existing try/except, so a malformed last_sync will raise ValueError and crash before the error handling is reached. Please wrap the parse in its own try/except and treat invalid timestamps as None, e.g.:
device_dt: datetime | None = None
if device.last_sync:
try:
device_dt = isoparse(device.last_sync)
except ValueError:
self.log(message=f"Invalid last_sync on device {device.uid}: {device.last_sync}", level="warning")This keeps the checkpoint filter resilient to bad timestamp data.
Suggested change
| for page in self._fetch_device_pages(): | |
| for device in page.results: | |
| device_dt = isoparse(device.last_sync) if device.last_sync else None | |
| # Client-side checkpoint filter: skip devices not modified since last run. | |
| for page in self._fetch_device_pages(): | |
| for device in page.results: | |
| device_dt: datetime | None = None | |
| if device.last_sync: | |
| try: | |
| device_dt = isoparse(device.last_sync) | |
| except ValueError: | |
| self.log( | |
| message=f"Invalid last_sync on device {device.uid}: {device.last_sync}", | |
| level="warning", | |
| ) | |
| # Client-side checkpoint filter: skip devices not modified since last run. |
Comment on lines
+104
to
+107
| def test_operating_system_unknown_family(connector): | ||
| device = connector.build_device(HolmDevice.model_validate(make_device(os_family="plan9"))) | ||
| assert device.os.type == OSTypeStr.OTHER | ||
| assert device.os.type_id == OSTypeId.OTHER |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add tests for build_operating_system handling of mixed-case and whitespace os_family values.
Current tests only cover lowercase os_family values and an unknown family, but the implementation normalizes via strip().lower(). Please add cases like os_family=" Windows " and "MAC" to assert they resolve to the correct OSTypeStr/OSTypeId, so the normalization behavior is explicitly covered.
Suggested implementation:
def test_operating_system_windows(connector):
device = connector.build_device(HolmDevice.model_validate(make_device()))
assert device.os is not None
assert device.os.name == "Microsoft Windows Server 2025 Datacenter"
assert device.os.type == OSTypeStr.WINDOWS
assert device.os.type_id == OSTypeId.WINDOWS
def test_operating_system_windows_mixed_case_whitespace(connector):
device = connector.build_device(
HolmDevice.model_validate(make_device(os_family=" Windows "))
)
assert device.os is not None
assert device.os.type == OSTypeStr.WINDOWS
assert device.os.type_id == OSTypeId.WINDOWS
def test_operating_system_mac_mixed_case(connector):
device = connector.build_device(
HolmDevice.model_validate(make_device(os_family="MAC"))
)
assert device.os is not None
assert device.os.type == OSTypeStr.MAC_OS
assert device.os.type_id == OSTypeId.MAC_OS
import pytest- Verify the exact enum members for macOS in
OSTypeStrandOSTypeId(e.g. they might beMAC_OS,MAC_OS_X, or similar) and adjust the assertions intest_operating_system_mac_mixed_caseaccordingly. - If there is a dedicated factory or fixture for macOS devices (rather than using
make_device(os_family="MAC")), update the test to use it to stay consistent with the rest of the test suite.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Relates to 1774, 1772
Summary by Sourcery
Add a Holm Security automation module that authenticates against the Holm Security API, validates credentials, and exposes a device asset connector producing OCSF Device Inventory Info events from /v2/devices.
New Features:
Enhancements:
Build:
Documentation:
Tests: