v0.2.12

deployments/helm/openhound/README.md

@@ -46,11 +46,11 @@ collector:
   name: jamf
   extraSecretMounts: { }
 
-# The BHE destination config. This is typically stored inside the same secrets.toml referenced by config.existingSecret, but non-sensitive values can be provided here for convenience.
+# The BHE destination url is non-sensitive config; set it here, via environment variable, or in config.toml.
+# Sensitive credentials (token_id and token_key) belong in secrets.toml referenced by config.existingSecret.
 destination:
   bloodhoundEnterprise:
     url: https://test.bloodhoundenterprise.io
-    interval: "300"
 
 ```
 

deployments/helm/values.example.yaml

@@ -21,7 +21,8 @@ collector:
   name: jamf
   extraSecretMounts: { }
 
-# The BHE destination config. This is typically stored inside the same secrets.toml referenced by config.existingSecret, but non-sensitive values can be provided here for convenience.
+# The BHE destination url is non-sensitive config; set it here, via environment variable, or in config.toml.
+# Sensitive credentials (token_id and token_key) belong in secrets.toml referenced by config.existingSecret.
 destination:
   bloodhoundEnterprise:
     url: https://test.bloodhoundenterprise.io

example-configurations/bloodhound-enterprise/.dlt-example/config.toml

@@ -14,6 +14,10 @@ request_max_retry_delay = 900
 # To switch to structured JSON instead, uncomment the line below (must be uppercase "JSON")
 # log_format = "JSON"
 
+# BHE destination URL (non-sensitive; tokens go in secrets.toml)
+[destination.bloodhoundenterprise]
+url = "bhe_url"
+
 [extract]
 workers = 8
 

example-configurations/bloodhound-enterprise/.dlt-example/secrets_github.toml

@@ -1,7 +1,6 @@
 [destination.bloodhoundenterprise]
 token_id = "client_token_id"
 token_key = "client_token_key"
-url = "bhe_url"
 
 # Example configuration for github secrets: https://bloodhound.specterops.io/openhound/collectors/github/collect-data#example-configuration
 [sources.source.github.credentials]

example-configurations/bloodhound-enterprise/.dlt-example/secrets_jamf.toml

@@ -1,7 +1,6 @@
 [destination.bloodhoundenterprise]
 token_id = "client_token_id"
 token_key = "client_token_key"
-url = "bhe_url"
 
 # Example configuration for jamf secrets: https://bloodhound.specterops.io/openhound/collectors/jamf/collect-data#example-configuration
 [sources.source.jamf.credentials]

example-configurations/bloodhound-enterprise/.dlt-example/secrets_okta.toml

@@ -1,7 +1,6 @@
 [destination.bloodhoundenterprise]
 token_id = "client_token_id"
 token_key = "client_token_key"
-url = "bhe_url"
 
 # Example configuration for okta secrets: https://bloodhound.specterops.io/openhound/collectors/okta/collect-data#example-configuration
 [sources.source.okta.credentials]

example-configurations/bloodhound-enterprise/docker-compose.yml

@@ -50,7 +50,7 @@ services:
 
 secrets:
   # Copy the .dlt-example folder to ${HOME}/.dlt as a starting point for each secrets file.
-  # Each secrets file must also contain [destination.bloodhoundenterprise] with url, token_id, and token_key.
+  # Each secrets file must contain [destination.bloodhoundenterprise] with token_id and token_key.
 
   # Jamf: username + password auth
   secrets_jamf:

src/openhound/core/logging.py

@@ -242,6 +242,10 @@ def __init__(
         self.base_path = Path(base_path) if base_path else self.default_platform_path()
         self.log_file_path: Path | None = None
 
+        # Share one rotating handler per file across loggers; opening the same file
+        # with multiple handlers breaks rotation on Windows (WinError 32).
+        self._file_handlers: dict[Path, "RotatingFileHandler"] = {}
+
         self.handlers = {
             LogMode.CLI: self.cli_handlers,
             LogMode.CONTAINER: self.container_handlers,
@@ -374,33 +378,45 @@ def _file_formatter(self) -> logging.Formatter:
             return OpenHoundJSONFormatter()
         return OpenHoundTextFormatter()
 
-    def container_handlers(self, logger: logging.Logger, file_path: Path) -> None:
-        """Set the logging handler/format when running in a container"""
-
-        formatter = self._file_formatter()
-
-        # Log to stdout for better compatibility with container-based logging systems.
-        # Output is human-readable text by default; set runtime.log_format = "JSON" for structured JSON.
-        stdout_handler = logging.StreamHandler(sys.stdout)
-        stdout_handler.setFormatter(formatter)
-        logger.addHandler(stdout_handler)
-
-        # But also log the same format to a file for persistence and debugging when needed
+    def _build_file_handler(self, file_path: Path) -> "RotatingFileHandler":
+        """Create and configure a rotating file handler for the given path."""
         rotating_file_handler = RotatingFileHandler(
             file_path,
             when=self.rotate_when,
             interval=self.interval,
             backupCount=self.backup_count,
             max_bytes=self.max_bytes,
         )
-        rotating_file_handler.setFormatter(formatter)
+        rotating_file_handler.setFormatter(self._file_formatter())
         # This regular expression overrides the default extMatch to recognize both
         # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well)
         rotating_file_handler.extMatch = re.compile(
             r"(?<!\d)\d{4}-\d{2}-\d{2}_\d{2}(-\d{2}-\d{2})?(?!\d)", re.ASCII
         )
+        return rotating_file_handler
 
-        logger.addHandler(rotating_file_handler)
+    def _get_file_handler(self, file_path: Path) -> "RotatingFileHandler":
+        """Return a shared rotating file handler for the path (one open file per path)."""
+        key = Path(file_path).resolve()
+        handler = self._file_handlers.get(key)
+        if handler is None:
+            handler = self._build_file_handler(file_path)
+            self._file_handlers[key] = handler
+        return handler
+
+    def container_handlers(self, logger: logging.Logger, file_path: Path) -> None:
+        """Set the logging handler/format when running in a container"""
+
+        formatter = self._file_formatter()
+
+        # Log to stdout for better compatibility with container-based logging systems.
+        # Output is human-readable text by default; set runtime.log_format = "JSON" for structured JSON.
+        stdout_handler = logging.StreamHandler(sys.stdout)
+        stdout_handler.setFormatter(formatter)
+        logger.addHandler(stdout_handler)
+
+        # But also log the same format to a file for persistence and debugging when needed
+        logger.addHandler(self._get_file_handler(file_path))
 
     def cli_handlers(self, logger: logging.Logger, file_path: Path) -> None:
         """Set the logging handler/format when running as a standalone CLI tool"""
@@ -420,40 +436,11 @@ def cli_handlers(self, logger: logging.Logger, file_path: Path) -> None:
         logger.addHandler(console_handler)
 
         # But also save the logs to a file using the configured format (text by default)
-        file_formatter = self._file_formatter()
-        rotating_file_handler = RotatingFileHandler(
-            file_path,
-            when=self.rotate_when,
-            interval=self.interval,
-            backupCount=self.backup_count,
-            max_bytes=self.max_bytes,
-        )
-        rotating_file_handler.setFormatter(file_formatter)
-        # This regular expression overrides the default extMatch to recognize both
-        # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well)
-        rotating_file_handler.extMatch = re.compile(
-            r"(?<!\d)\d{4}-\d{2}-\d{2}_\d{2}(-\d{2}-\d{2})?(?!\d)", re.ASCII
-        )
-
-        logger.addHandler(rotating_file_handler)
+        logger.addHandler(self._get_file_handler(file_path))
 
     def service_handlers(self, logger: logging.Logger, file_path: Path) -> None:
         """Set the logging handler/format when running the OpenHound service"""
-        file_formatter = self._file_formatter()
-        rotating_file_handler = RotatingFileHandler(
-            file_path,
-            when=self.rotate_when,
-            interval=self.interval,
-            backupCount=self.backup_count,
-            max_bytes=self.max_bytes,
-        )
-        rotating_file_handler.setFormatter(file_formatter)
-        # This regular expression overrides the default extMatch to recognize both
-        # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well)
-        rotating_file_handler.extMatch = re.compile(
-            r"(?<!\d)\d{4}-\d{2}-\d{2}_\d{2}(-\d{2}-\d{2})?(?!\d)", re.ASCII
-        )
-        logger.addHandler(rotating_file_handler)
+        logger.addHandler(self._get_file_handler(file_path))
 
     @property
     def runtime_mode(self) -> LogMode:

src/openhound/core/pipeline.py

@@ -1,3 +1,7 @@
+import errno
+import logging
+import sys
+import time
 from abc import ABC, abstractmethod
 
 from dlt.common.configuration.exceptions import ConfigFieldMissingException
@@ -8,33 +12,80 @@
 
 from openhound.core.exceptions import ConfigException, ParseException
 
+logger = logging.getLogger(__name__)
+
+# Windows can transiently lock freshly written load-package files, so dlt's
+# per-file rename fallback raises PermissionError (WinError 5); retry the run.
+_TRANSIENT_RENAME_ERRNOS = {errno.EACCES, errno.EPERM}
+_MAX_TRANSIENT_RETRIES = 5
+_TRANSIENT_RETRY_BACKOFF = 0.25
+
+
+def _transient_filesystem_cause(err: BaseException) -> OSError | None:
+    """Return the underlying transient filesystem error in the cause chain, if any."""
+    seen: set[int] = set()
+    current: BaseException | None = err
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        if isinstance(current, OSError) and current.errno in _TRANSIENT_RENAME_ERRNOS:
+            return current
+        nested = getattr(current, "exception", None)
+        if nested is None:
+            nested = current.__cause__ or current.__context__
+        current = nested
+    return None
+
 
 class BasePipeline(ABC):
     @property
     @abstractmethod
     def pipeline(self) -> "Pipeline": ...
 
     def _run(self, source, **kwargs) -> LoadInfo:
-        try:
-            return self.pipeline.run(source, **kwargs)
-        except PipelineStepFailed as err:
-            if isinstance(err.exception, ConfigFieldMissingException):
-                config_cause: ConfigFieldMissingException = err.exception
-                raise ConfigException(
-                    pipeline_name=err.pipeline.pipeline_name,
-                    destination=str(err.pipeline.destination),
-                    dataset_name=err.pipeline.dataset_name,
-                    spec_name=config_cause.spec_name,
-                    message=str(err.exception),
-                ) from None
-
-            if isinstance(err.exception, ResourceExtractionError):
-                extract_cause: ResourceExtractionError = err.exception
-                raise ParseException(
-                    pipeline_name=err.pipeline.pipeline_name,
-                    destination=str(err.pipeline.destination),
-                    dataset_name=err.pipeline.dataset_name,
-                    step=extract_cause.pipe_name,
-                    message=extract_cause.msg,
-                )
-            raise
+        last_err: BaseException | None = None
+        for attempt in range(_MAX_TRANSIENT_RETRIES):
+            try:
+                return self.pipeline.run(source, **kwargs)
+            except PipelineStepFailed as err:
+                if isinstance(err.exception, ConfigFieldMissingException):
+                    config_cause: ConfigFieldMissingException = err.exception
+                    raise ConfigException(
+                        pipeline_name=err.pipeline.pipeline_name,
+                        destination=str(err.pipeline.destination),
+                        dataset_name=err.pipeline.dataset_name,
+                        spec_name=config_cause.spec_name,
+                        message=str(err.exception),
+                    ) from None
+
+                if isinstance(err.exception, ResourceExtractionError):
+                    extract_cause: ResourceExtractionError = err.exception
+                    raise ParseException(
+                        pipeline_name=err.pipeline.pipeline_name,
+                        destination=str(err.pipeline.destination),
+                        dataset_name=err.pipeline.dataset_name,
+                        step=extract_cause.pipe_name,
+                        message=extract_cause.msg,
+                    )
+
+                if _transient_filesystem_cause(err) is None:
+                    raise
+                last_err = err
+            except PermissionError as err:
+                if sys.platform != "win32" or err.errno not in _TRANSIENT_RENAME_ERRNOS:
+                    raise
+                last_err = err
+
+            if attempt + 1 >= _MAX_TRANSIENT_RETRIES:
+                break
+
+            logger.warning(
+                "Transient filesystem error; retrying run (%d/%d): %s",
+                attempt + 1,
+                _MAX_TRANSIENT_RETRIES,
+                last_err,
+            )
+            time.sleep(_TRANSIENT_RETRY_BACKOFF * (attempt + 1))
+
+        if last_err is None:
+            raise RuntimeError("unreachable: retry loop exited without an error")
+        raise last_err

src/openhound/destinations/bloodhound_enterprise/destination.py

@@ -16,7 +16,7 @@
 def ingest(
     items: TDataItems,
     table: TTableSchema,
-    url: str = dlt.secrets.value,
+    url: str = dlt.config.value,
     token_id: str = dlt.secrets.value,
     token_key: str = dlt.secrets.value,
     source_kind: str = dlt.config.value,