[sigmaHQ] New connector to ingest Sigma rules #5525
Conversation
romain-filigran
added
new
There was a problem hiding this comment.
Pull request overview
This PR introduces a new external import connector for ingesting Sigma detection rules from the SigmaHQ repository into OpenCTI as indicators. The connector fetches rule packages from GitHub releases, converts them to STIX format, and creates indicators with relationships to attack patterns and vulnerabilities.
Key changes:
- New connector implementation with GitHub API client to fetch Sigma rule packages
- STIX conversion logic that transforms Sigma rules into indicators with relationships
- Configuration support for multiple rule package types (all rules, core, core+, core++, emerging threats)
Reviewed changes
Copilot reviewed 16 out of 17 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| external-import/sigmahq/src/sigmahq_client/api_client.py | GitHub API client for downloading and parsing Sigma rule packages |
| external-import/sigmahq/src/connector/converter_to_stix.py | Converts Sigma rules to STIX indicators with attack pattern and vulnerability relationships |
| external-import/sigmahq/src/connector/connector.py | Main connector orchestration logic with state management |
| external-import/sigmahq/src/connector/settings.py | Configuration model for connector and SigmaHQ-specific settings |
| external-import/sigmahq/src/requirements.txt | Python dependencies including pycti, pydantic, and pySigma |
| external-import/sigmahq/README.md | Documentation describing connector purpose and installation |
| external-import/sigmahq/Dockerfile | Container image definition for the connector |
| external-import/sigmahq/docker-compose.yml | Docker compose configuration example |
| external-import/sigmahq/config.yml.sample | Sample configuration file |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # Define headers in session and update when needed | ||
| self.session = requests.Session() | ||
|
|
||
| def get_lastest_published_version(self) -> dict[str, Any] | None: |
There was a problem hiding this comment.
Corrected spelling of 'lastest' to 'latest'.
| stix_objects.extend(stix_entities) | ||
| except Exception as err: | ||
| self.helper.connector_logger.error( | ||
| f"An exception occurred while converting SigmaHQ rule: {rule.filename}", |
There was a problem hiding this comment.
The variable rule is a dictionary with keys 'filename' and 'rule_content', not an object with a .filename attribute. This will raise an AttributeError. Change to rule['filename'] or rule.get('filename').
| f"An exception occurred while converting SigmaHQ rule: {rule.filename}", | |
| f"An exception occurred while converting SigmaHQ rule: {rule.get('filename')}", |
| ) | ||
|
|
||
| # get latest rule package version | ||
| release_metadata = self.client.get_lastest_published_version() |
There was a problem hiding this comment.
Corrected spelling of 'get_lastest_published_version' to 'get_latest_published_version'.
| release_metadata = self.client.get_lastest_published_version() | |
| release_metadata = self.client.get_latest_published_version() |
| rules = self.client.download_and_convert_package( | ||
| asset["browser_download_url"] | ||
| ) | ||
|
|
There was a problem hiding this comment.
If no matching asset is found in the loop at lines 35-39, rules will remain None, causing a TypeError when attempting to iterate. Add a check after the asset loop to handle the case where rules is still None.
| if rules is None: | |
| self.helper.connector_logger.warning( | |
| f"No matching asset found for rule package '{rule_package}'." | |
| ) | |
| return stix_objects |
| def download_and_convert_package(self, url): | ||
|
|
There was a problem hiding this comment.
Missing docstring for the download_and_convert_package method. Should document the expected format of the url parameter and describe the return value structure (list of dictionaries with 'filename' and 'rule_content' keys).
| def download_and_convert_package(self, url): | |
| def download_and_convert_package(self, url): | |
| """ | |
| Download a Sigma rule package from a ZIP URL and convert its YAML files. | |
| The given URL is expected to be an HTTP(S) URL pointing to a ZIP archive | |
| that contains Sigma rule files. Only files with a ``.yml`` extension are | |
| processed; directories and other file types inside the archive are ignored. | |
| :param url: HTTP or HTTPS URL of the ZIP archive containing Sigma rules. | |
| :return: A list of dictionaries, each representing a Sigma rule with: | |
| - ``filename``: the path/name of the rule file inside the archive. | |
| - ``rule_content``: the UTF-8 decoded text content of the rule file. | |
| Returns ``None`` if an error occurs during download or extraction. | |
| """ |
Proposed changes
Related issues
Checklist
Further comments