Feature: Holm Security Vulnerabilities#2915
Open
vg-svitla wants to merge 1 commit into
Open
Conversation
Contributor
Reviewer's GuideIntroduce a Holm Security vulnerability asset connector that maps vulnerability findings to OCSF, refactor shared device mapping logic into reusable helpers, and tighten account validation by requiring both device and vulnerability endpoints to succeed, with comprehensive tests and manifest/changelog updates. Sequence diagram for tightened Holm Security account validationsequenceDiagram
actor User
participant HolmSecurityAccountValidator as HolmSecurityAccountValidator
participant ApiClient as ApiClient
participant HolmSecurityAPI as HolmSecurityAPI
User->>HolmSecurityAccountValidator: validate()
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: log("Starting credentials validation", info)
loop VALIDATION_ENDPOINTS
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: _check_endpoint(endpoint, params)
HolmSecurityAccountValidator->>ApiClient: client.get(base_url + endpoint, params, timeout)
alt success
ApiClient-->>HolmSecurityAccountValidator: Response(status_code=200)
HolmSecurityAccountValidator-->>HolmSecurityAccountValidator: return True
else non_200
ApiClient-->>HolmSecurityAccountValidator: Response(status_code!=200)
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: _describe_error(url, response)
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: log(error_message, error)
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: error(error_message)
HolmSecurityAccountValidator-->>HolmSecurityAccountValidator: return False
end
end
alt all endpoints succeeded
HolmSecurityAccountValidator->>HolmSecurityAccountValidator: log("credentials validated successfully", info)
HolmSecurityAccountValidator-->>User: True
else some endpoint failed
HolmSecurityAccountValidator-->>User: False
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![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 3 issues, and left some high level feedback:
- In
get_assets, the direct use ofisoparseonfinding.last_detectedand the checkpoint value can cause the whole run to fail on a single malformed timestamp; consider using a safer parser (similar tomappers.to_epoch) with error handling so bad records are skipped rather than aborting the connector. - The
_fetch_pageshelper assumes the Holm API always returns a well-formed JSON structure for pagination and immediately callsresponse.json()andmodel_validate; adding small defensive checks or logging around unexpected payload shapes (e.g. missingnextorresults) would make failures easier to diagnose without relying solely on pydantic validation errors.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `get_assets`, the direct use of `isoparse` on `finding.last_detected` and the checkpoint value can cause the whole run to fail on a single malformed timestamp; consider using a safer parser (similar to `mappers.to_epoch`) with error handling so bad records are skipped rather than aborting the connector.
- The `_fetch_pages` helper assumes the Holm API always returns a well-formed JSON structure for pagination and immediately calls `response.json()` and `model_validate`; adding small defensive checks or logging around unexpected payload shapes (e.g. missing `next` or `results`) would make failures easier to diagnose without relying solely on pydantic validation errors.
## Individual Comments
### Comment 1
<location path="HolmSecurity/holm_security/asset_connector/vulnerability_assets.py" line_range="355-361" />
<code_context>
+ 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")
</code_context>
<issue_to_address>
**issue (bug_risk):** Initialize `_latest_time` to avoid AttributeError in `update_checkpoint` when no findings are processed.
`update_checkpoint` assumes `self._latest_time` exists, but it’s only set in `get_assets` when `max_raw` changes. If there are no findings or all are skipped before `_latest_time` is set, `update_checkpoint` will raise `AttributeError`. Initialize `self._latest_time: str | None = None` in `__init__`, or use `getattr(self, "_latest_time", None)` inside `update_checkpoint` to avoid this failure.
</issue_to_address>
### Comment 2
<location path="HolmSecurity/holm_security/asset_connector/vulnerability_assets.py" line_range="324-325" />
<code_context>
+ # 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.
</code_context>
<issue_to_address>
**issue (bug_risk):** Use a safer timestamp parsing strategy for `last_detected` to avoid aborting the run on malformed data.
In `get_assets`, `finding_dt` is derived via `isoparse(finding.last_detected)`. If Holm returns a malformed timestamp, this will raise and stop the whole connector run. Consider either reusing `mappers.to_epoch` (then converting to `datetime`) or wrapping `isoparse` in try/except and treating unparsable values as `None` so a single bad record doesn’t break ingestion.
</issue_to_address>
### Comment 3
<location path="HolmSecurity/holm_security/asset_connector/vulnerability_assets.py" line_range="165-179" />
<code_context>
+ 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),
</code_context>
<issue_to_address>
**suggestion:** Prefer the host name from the finding when populating network interface hostname.
In `build_device_from_cache`, `network_interfaces` are built with `cached.hostname`, even when `host_name` is available. If the cached device has no hostname but the finding does, interface hostnames are lost. Use `host_name` (computed earlier) instead of `cached.hostname` when calling `build_network_interfaces` to keep interface hostnames aligned with the resolved device fields.
```suggestion
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, host_name or cached.hostname),
first_seen_time=mappers.to_epoch(cached.created),
last_seen_time=mappers.to_epoch(cached.last_sync),
```
</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
+355
to
+361
| 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") |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Initialize _latest_time to avoid AttributeError in update_checkpoint when no findings are processed.
update_checkpoint assumes self._latest_time exists, but it’s only set in get_assets when max_raw changes. If there are no findings or all are skipped before _latest_time is set, update_checkpoint will raise AttributeError. Initialize self._latest_time: str | None = None in __init__, or use getattr(self, "_latest_time", None) inside update_checkpoint to avoid this failure.
Comment on lines
+324
to
+325
| for finding in page.results: | ||
| finding_dt = isoparse(finding.last_detected) if finding.last_detected else None |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Use a safer timestamp parsing strategy for last_detected to avoid aborting the run on malformed data.
In get_assets, finding_dt is derived via isoparse(finding.last_detected). If Holm returns a malformed timestamp, this will raise and stop the whole connector run. Consider either reusing mappers.to_epoch (then converting to datetime) or wrapping isoparse in try/except and treating unparsable values as None so a single bad record doesn’t break ingestion.
Comment on lines
+165
to
+179
| 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), |
Contributor
There was a problem hiding this comment.
suggestion: Prefer the host name from the finding when populating network interface hostname.
In build_device_from_cache, network_interfaces are built with cached.hostname, even when host_name is available. If the cached device has no hostname but the finding does, interface hostnames are lost. Use host_name (computed earlier) instead of cached.hostname when calling build_network_interfaces to keep interface hostnames aligned with the resolved device fields.
Suggested change
| 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), | |
| 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, host_name or cached.hostname), | |
| first_seen_time=mappers.to_epoch(cached.created), | |
| last_seen_time=mappers.to_epoch(cached.last_sync), |
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 1620, 1517
Summary by Sourcery
Introduce a Holm Security vulnerability asset connector that maps vulnerability findings to OCSF and aligns device mapping via shared helpers, while updating account validation and module wiring accordingly.
New Features:
Enhancements:
Build:
Documentation:
Tests: