fix(infrastructure): SecretStr serialisation + logger + scheduler correctness

- Configuration adapter: SecretStr serialisation for dumps
- Bearer-token auth: constant-time compare + ASCII guard on candidate
- Logger: file-handler for audit sink when audit_log_file configured
- Scheduler: response formatter default spreads input dict; docstring fix
- Template repository: safe deserialize path
- Utilities: collection validation + json_utils typing

Author: fgogolli

src/orb/infrastructure/adapters/configuration_adapter.py

@@ -20,8 +20,13 @@ def __init__(self, config_manager: ConfigurationManager, logger: LoggingPort) ->
         self._logger = logger
 
     def get_app_config(self) -> dict[str, Any]:
-        """Get structured application configuration."""
-        return self._config_manager.app_config.model_dump()
+        """Get structured application configuration.
+
+        Uses ``mode="json"`` so pydantic types like ``SecretStr`` serialise to
+        masked strings instead of opaque Python objects — keeps the result
+        JSON-serialisable for REST callers (e.g. the Config UI).
+        """
+        return self._config_manager.app_config.model_dump(mode="json")
 
     @property
     def app_config(self) -> "AppConfig":
@@ -124,6 +129,11 @@ def get_template_config(self) -> dict[str, Any]:
             self._logger.warning("Failed to get template config: %s", e)
             return {}
 
+    def get_raw_config(self) -> dict[str, Any]:
+        """Return the raw on-disk configuration dict before Pydantic hydration."""
+        raw = self._config_manager._ensure_raw_config()
+        return dict(raw) if isinstance(raw, dict) else {}
+
     def get_metrics_config(self) -> dict[str, Any]:
         """Get metrics configuration."""
 
@@ -146,7 +156,7 @@ def get_metrics_config(self) -> dict[str, Any]:
 
         try:
             # Get metrics section from raw config
-            raw = self._config_manager._ensure_raw_config()  # type: ignore[attr-defined]
+            raw = self.get_raw_config()
             metrics_config = raw.get("metrics", {}) if isinstance(raw, dict) else {}
 
             result: dict[str, Any] = defaults.copy()
@@ -173,6 +183,22 @@ def get_loaded_config_file(self) -> str | None:
         """Get the path of the loaded configuration file."""
         return self._config_manager.get_loaded_config_file()
 
+    def save_config(self, path: str | None = None) -> str:
+        """Persist the in-memory raw config to disk.
+
+        If ``path`` is None, writes to the currently-loaded config file.
+        Returns the resolved path that was written. Raises if no path can
+        be resolved (e.g. config came from env only, no file backing).
+        """
+        target = path or self._config_manager.get_loaded_config_file()
+        if not target:
+            raise ValueError(
+                "No config file path resolved — config was not loaded from a file. "
+                "Pass an explicit path to save_config()."
+            )
+        self._config_manager.save(target)
+        return target
+
     def get_root_dir(self) -> str:
         """Get the root directory path."""
         from orb.config.platform_dirs import get_root_location

src/orb/infrastructure/auth/strategy/bearer_token_strategy.py

@@ -223,7 +223,17 @@ def from_auth_config(cls, auth_config: Any) -> BearerTokenStrategy:
                 "Bearer token authentication requires auth.bearer_token configuration. "
                 "Set auth.bearer_token.secret_key in your server config."
             )
-        secret_key: str = getattr(bearer_cfg, "secret_key", "") or ""
+        raw_secret = getattr(bearer_cfg, "secret_key", None)
+        if raw_secret is None:
+            raise ConfigurationError(
+                "Bearer token authentication requires a secret_key in auth.bearer_token config."
+            )
+        # SecretStr — call .get_secret_value() to obtain the plain string for JWT ops.
+        secret_key: str = (
+            raw_secret.get_secret_value()
+            if hasattr(raw_secret, "get_secret_value")
+            else str(raw_secret)
+        )
         if not secret_key:
             raise ConfigurationError(
                 "Bearer token authentication requires a secret_key in auth.bearer_token config."

src/orb/infrastructure/auth/strategy/bearer_token_strategy_enhanced.py

@@ -349,7 +349,17 @@ def from_auth_config(cls, auth_config: Any) -> EnhancedBearerTokenStrategy:
             raise ConfigurationError(
                 "Enhanced bearer token authentication requires auth.bearer_token configuration."
             )
-        secret_key: str = getattr(bearer_cfg, "secret_key", "") or ""
+        raw_secret = getattr(bearer_cfg, "secret_key", None)
+        if raw_secret is None:
+            raise ConfigurationError(
+                "Enhanced bearer token authentication requires a secret_key in auth.bearer_token config."
+            )
+        # SecretStr — call .get_secret_value() to obtain the plain string for JWT ops.
+        secret_key: str = (
+            raw_secret.get_secret_value()
+            if hasattr(raw_secret, "get_secret_value")
+            else str(raw_secret)
+        )
         if not secret_key:
             raise ConfigurationError(
                 "Enhanced bearer token authentication requires a secret_key in auth.bearer_token config."

src/orb/infrastructure/error/decorators.py

@@ -9,6 +9,17 @@
 from functools import wraps
 from typing import Any, Callable, Optional
 
+try:
+    from fastapi import HTTPException as _FastAPIHTTPException  # type: ignore[assignment]
+
+    _HTTP_EXCEPTION_TYPE: type[BaseException] = _FastAPIHTTPException  # type: ignore[assignment]
+except ImportError:  # FastAPI is optional — define a private sentinel never raised at runtime
+
+    class _FastAPIHTTPException(Exception):  # type: ignore[no-redef]
+        """Placeholder so the except clause is valid when FastAPI is absent."""
+
+    _HTTP_EXCEPTION_TYPE = _FastAPIHTTPException
+
 from orb.infrastructure.error.exception_handler import (
     ExceptionContext,
     ExceptionHandler,
@@ -54,6 +65,8 @@ def decorator(func: Callable) -> Callable:
             async def async_wrapper(*args, **kwargs):
                 try:
                     return await func(*args, **kwargs)
+                except _HTTP_EXCEPTION_TYPE:
+                    raise
                 except Exception as e:
                     # Get exception handler (use provided or global singleton)
                     exception_handler = handler or get_exception_handler()
@@ -83,6 +96,8 @@ def sync_wrapper(*args, **kwargs):
                 """Synchronous wrapper with error handling."""
                 try:
                     return func(*args, **kwargs)
+                except _HTTP_EXCEPTION_TYPE:
+                    raise
                 except Exception as e:
                     # Get exception handler (use provided or global singleton)
                     exception_handler = handler or get_exception_handler()

src/orb/infrastructure/logging/logger.py

@@ -293,11 +293,67 @@ def debug(self, msg: str, **kwargs: Any) -> None:
         self.logger.debug(msg, extra=kwargs)
 
 
+def setup_audit_logger(audit_log_file: Optional[str] = None) -> None:
+    """Attach a dedicated handler to the ``orb.audit`` logger.
+
+    In container deployments with stdout-only logging the audit logger
+    inherits the root handlers and that is sufficient.  When operators want
+    audit records in a separate, structured file (for SIEM ingestion, long-term
+    retention, or audit trail requirements) they set ``server.audit_log_file``
+    in config and this function attaches a ``RotatingFileHandler`` that writes
+    one JSON object per line.
+
+    If ``audit_log_file`` is ``None`` a structured ``StreamHandler`` (stderr)
+    is attached instead so audit records always have at least one dedicated
+    channel with JSON formatting, distinct from the coloured console output.
+
+    This function is idempotent — calling it more than once with the same path
+    does not add duplicate handlers.
+
+    Args:
+        audit_log_file: Absolute path to the audit log file.  ``None`` means
+            write structured JSON to stderr.
+    """
+    audit_log = logging.getLogger("orb.audit")
+    # Prevent audit records from propagating to the root logger when a
+    # dedicated handler is configured — keeps the two streams independent.
+    audit_log.propagate = False
+
+    json_fmt = JsonFormatter(log_type="audit")
+
+    if audit_log_file:
+        log_path = Path(audit_log_file)
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        # Check for an existing handler pointing at the same file to stay
+        # idempotent across multiple create_fastapi_app calls in tests.
+        existing_paths = {
+            getattr(h, "baseFilename", None)
+            for h in audit_log.handlers
+            if isinstance(h, logging.handlers.RotatingFileHandler)
+        }
+        if str(log_path.resolve()) not in existing_paths:
+            fh = logging.handlers.RotatingFileHandler(
+                filename=str(log_path),
+                maxBytes=50 * 1024 * 1024,  # 50 MiB per shard
+                backupCount=10,
+                encoding="utf-8",
+            )
+            fh.setFormatter(json_fmt)
+            audit_log.addHandler(fh)
+    # Structured stderr handler — only add if no handlers exist yet.
+    elif not audit_log.handlers:
+        import sys
+
+        sh = logging.StreamHandler(sys.stderr)
+        sh.setFormatter(json_fmt)
+        audit_log.addHandler(sh)
+
+
 class AuditLogger:
     """Logger for audit events."""
 
     def __init__(self) -> None:
-        self.logger = get_logger("audit")
+        self.logger = get_logger("orb.audit")
 
     def log_event(
         self,

src/orb/infrastructure/scheduler/default/default_strategy.py

@@ -266,35 +266,28 @@ def _serialize_machine(self, machine: Any) -> dict[str, Any]:
         return d
 
     def format_machine_details_response(self, machine_data: dict) -> dict:
-        """Format machine details with default fields, including provider_data fields."""
+        """Format machine details for the API detail endpoint.
+
+        Returns every field on the source dict so adding a domain field to
+        ``Machine`` automatically surfaces in the detail payload — no
+        formatter update needed. ``id`` is kept as a legacy alias for
+        ``machine_id``; ``provider`` is hard-stamped because the default
+        strategy is the bound provider.
+        """
         provider_data: dict[str, Any] = machine_data.get("provider_data") or {}
+        machine_id = machine_data.get("machine_id") or machine_data.get("id")
 
-        result: dict[str, Any] = {
-            "id": machine_data.get("id"),
-            "name": machine_data.get("name"),
-            "status": machine_data.get("status"),
-            "provider": "default",
-            "instance_type": machine_data.get("instance_type"),
-            "image_id": machine_data.get("image_id"),
-            "private_ip": machine_data.get("private_ip"),
-            "public_ip": machine_data.get("public_ip"),
-            "subnet_id": machine_data.get("subnet_id"),
-            "security_group_ids": machine_data.get("security_group_ids"),
-            "status_reason": machine_data.get("status_reason"),
-            "launch_time": machine_data.get("launch_time"),
-            "termination_time": machine_data.get("termination_time"),
-            "tags": machine_data.get("tags"),
-        }
+        result: dict[str, Any] = {**machine_data}
+        result["id"] = machine_id
+        result["machine_id"] = machine_id
+        result["provider"] = "default"
+        result["provider_data"] = provider_data or None
 
-        # Provider_data fields — prefer top-level if already present, else pull from provider_data
+        # Promote selected provider_data sub-fields to the top level when the
+        # source dict did not already carry them — matches list endpoint shape.
         for key in ("region", "availability_zone", "vcpus", "health_checks"):
-            val = (
-                machine_data.get(key)
-                if machine_data.get(key) is not None
-                else provider_data.get(key)
-            )
-            if val is not None:
-                result[key] = val
+            if result.get(key) is None and provider_data.get(key) is not None:
+                result[key] = provider_data[key]
 
         cloud_host_id = machine_data.get("cloud_host_id") or provider_data.get("cloud_host_id")
         if cloud_host_id is not None:

src/orb/infrastructure/scheduler/hostfactory/response_formatter.py

@@ -1,11 +1,9 @@
 """HostFactory response formatting — converts domain objects to HF wire format.
 
-NOTE (dead code): HostFactoryResponseFormatter is not referenced anywhere in the
-production code path.  All equivalent formatting logic lives inline on
-HostFactorySchedulerStrategy (hostfactory_strategy.py).  This class was
-extracted as a refactoring step but the strategy was never updated to delegate
-to it.  Do not delete — it may be wired up in a future cleanup — but do not
-rely on it being called at runtime.
+This module is the contract-test target for HF response shapes
+(tests/unit/test_hf_response_contracts.py). HostFactorySchedulerStrategy
+keeps inline formatting; this class exists as the reference shape the
+tests assert against.
 """
 
 import json
@@ -178,6 +176,24 @@ def format_request_status_response(
             if req_dict.get("provider_api"):
                 hf_request["providerApi"] = req_dict["provider_api"]
 
+            # Lifecycle timestamps. HostFactory's wire spec is silent on
+            # these but our UI drawer + dashboard activity stepper rely on
+            # them. Forwarding when present keeps HF clients ignoring them
+            # (they're additional fields) while giving the UI what it needs.
+            for ts_field in (
+                "created_at",
+                "started_at",
+                "first_status_check",
+                "last_status_check",
+                "completed_at",
+                "request_type",
+                "successful_count",
+                "requested_count",
+                "template_id",
+            ):
+                if req_dict.get(ts_field) is not None:
+                    hf_request[ts_field] = req_dict[ts_field]
+
             formatted_requests.append(hf_request)
 
         return {"requests": formatted_requests}

src/orb/infrastructure/scheduler/registry.py

@@ -1,10 +1,13 @@
 """Scheduler Registry - Registry pattern for scheduler strategy factories."""
 
+import logging
 from typing import Any, Callable, ClassVar
 
 from orb.domain.base.exceptions import ConfigurationError
 from orb.infrastructure.registry.base_registry import BaseRegistration, BaseRegistry, RegistryMode
 
+logger = logging.getLogger(__name__)
+
 
 class UnsupportedSchedulerError(Exception):
     """Exception raised when an unsupported scheduler type is requested."""
@@ -134,8 +137,15 @@ def collect_defaults(self) -> dict:
                     defaults = reg.strategy_class.get_defaults_config()
                     if defaults:
                         self._deep_merge(merged, defaults)
-                except Exception:  # noqa: BLE001 — skip strategies that fail to load defaults
-                    pass
+                except Exception as exc:
+                    # One strategy's defaults loader is allowed to fail
+                    # without breaking the merge for the others. Log so
+                    # the bad strategy is observable.
+                    logger.warning(
+                        "Skipped defaults from %s: %s",
+                        getattr(reg.strategy_class, "__name__", reg),
+                        exc,
+                    )
         return merged
 
 

src/orb/infrastructure/services/iso_timestamp_service.py

@@ -1,15 +1,14 @@
 """Infrastructure implementation of timestamp service."""
 
 from datetime import datetime, timezone
-from typing import Union
 
 from orb.domain.services.timestamp_service import TimestampService
 
 
 class ISOTimestampService(TimestampService):
     """ISO format timestamp service implementation."""
 
-    def format_for_display(self, timestamp: Union[datetime, float, int, None]) -> str | None:
+    def format_for_display(self, timestamp: datetime | float | int | None) -> str | None:
         """Format timestamp to ISO format with Z timezone indicator."""
         if timestamp is None:
             return None
@@ -26,7 +25,7 @@ def format_for_display(self, timestamp: Union[datetime, float, int, None]) -> st
 
         return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
 
-    def format_for_dto(self, timestamp: Union[datetime, float, int, None]) -> int | None:
+    def format_for_dto(self, timestamp: datetime | float | int | None) -> int | None:
         """Format timestamp to unix timestamp for DTO backward compatibility."""
         if timestamp is None:
             return None
@@ -43,7 +42,7 @@ def current_timestamp(self) -> str:
         return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
     def format_with_type(
-        self, timestamp: Union[datetime, float, int, None], format_type: str
+        self, timestamp: datetime | float | int | None, format_type: str
     ) -> int | str | None:
         """Format timestamp based on requested format type."""
         if format_type == "unix":

src/orb/infrastructure/template/template_cache_service.py

@@ -3,14 +3,14 @@
 import threading
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
-from typing import Awaitable, Callable, Optional, Union
+from typing import Awaitable, Callable, Optional
 
 from orb.domain.base.ports import LoggingPort
 
 from .dtos import TemplateDTO
 
 # loader_func may return either a plain list or a coroutine
-LoaderFunc = Callable[[], Union[list[TemplateDTO], Awaitable[list[TemplateDTO]]]]
+LoaderFunc = Callable[[], list[TemplateDTO] | Awaitable[list[TemplateDTO]]]
 
 
 class TemplateCacheService(ABC):