Align instead of causing conflicts with nitlsconfig-python's audit logging and the driver runtimes

src/nitlsconfig/audit.py

@@ -11,10 +11,12 @@
 
 from __future__ import annotations
 
+import importlib
 import logging
 import sys
 import threading
 from enum import Enum
+from typing import Any
 
 from nitlsconfig._service import SERVICE_NAME
 from nitlsconfig.channel_tag import get_channel_target
@@ -37,6 +39,21 @@
 # below, because those do cross a trust boundary.
 
 _ROLE = "Client"
+# This source uses mscoree.dll as its EventMessageFile. Event ID 1000 selects its
+# generic literal-message template, which renders the complete first insertion
+# string as the event description. Other IDs can produce Event Viewer's
+# "message was not found in the message table" fallback instead.
+_WINDOWS_EVENT_ID = 1000
+
+# Match spdlog's Windows Event Log categories, which use its level enum values:
+# trace=0, debug=1, info=2, warn=3, err=4, critical=5.
+_WINDOWS_EVENT_CATEGORIES = {
+    logging.DEBUG: 1,
+    logging.INFO: 2,
+    logging.WARNING: 3,
+    logging.ERROR: 4,
+    logging.CRITICAL: 5,
+}
 
 
 class _TransportSecurity(Enum):
@@ -60,6 +77,65 @@ class _TransportSecurity(Enum):
 _ESCAPES = {"\\": "\\\\", "'": "\\'", "\r": "\\r", "\n": "\\n"}
 
 
+class _WindowsEventLogHandler(logging.Handler):
+    """Write to a pre-registered Windows Event Log source."""
+
+    def __init__(self, source_name: str) -> None:
+        super().__init__()
+        self._source_name = source_name
+        self._event_log: Any = importlib.import_module("win32evtlog")
+        self._user_sid = self._get_current_user_sid()
+        self._event_types = {
+            logging.DEBUG: self._event_log.EVENTLOG_INFORMATION_TYPE,
+            logging.INFO: self._event_log.EVENTLOG_INFORMATION_TYPE,
+            logging.WARNING: self._event_log.EVENTLOG_WARNING_TYPE,
+            logging.ERROR: self._event_log.EVENTLOG_ERROR_TYPE,
+            logging.CRITICAL: self._event_log.EVENTLOG_ERROR_TYPE,
+        }
+
+    @staticmethod
+    def _get_current_user_sid() -> Any:
+        """Return the current process token's user SID, or None if unavailable."""
+        try:
+            win32api = importlib.import_module("win32api")
+            win32con = importlib.import_module("win32con")
+            win32security = importlib.import_module("win32security")
+            token = win32security.OpenProcessToken(
+                win32api.GetCurrentProcess(), win32con.TOKEN_QUERY
+            )
+            try:
+                user_sid, _ = win32security.GetTokenInformation(token, win32security.TokenUser)
+                return user_sid
+            finally:
+                token.Close()
+        except Exception:
+            # A missing SID must not prevent the audit event itself from being recorded.
+            return None
+
+    def emit(self, record: logging.LogRecord) -> None:
+        """Emit one record without creating or changing Event Log registry keys."""
+        event_source = None
+        try:
+            event_source = self._event_log.RegisterEventSource(None, self._source_name)
+            event_type = self._event_types.get(record.levelno, self._event_log.EVENTLOG_ERROR_TYPE)
+            event_category = _WINDOWS_EVENT_CATEGORIES.get(
+                record.levelno,
+                4,  # Unknown/custom Python levels default to spdlog's error category.
+            )
+            self._event_log.ReportEvent(
+                event_source,
+                event_type,
+                event_category,
+                _WINDOWS_EVENT_ID,
+                self._user_sid,
+                [self.format(record)],
+                None,
+            )
+        finally:
+            if event_source is not None:
+                self._event_log.DeregisterEventSource(event_source)
+
+
 def _audit_field(value: object) -> str:
     """Return a bounded audit field with record-breaking characters escaped.
 
@@ -88,20 +164,24 @@ def _audit_field(value: object) -> str:
 def _make_logging_handler() -> logging.Handler:
     """Create the platform audit logging handler.
 
+    Windows uses the Windows Event Log and Linux uses syslog. Audit logging is
+    intentionally disabled on every other platform.
+
     Falls back to a null logging handler when the platform log is unreachable.
     That keeps audit records out of stderr, which ``logging`` would otherwise
     fall back to for a logger that has no logging handler of its own.
     """
     try:
         # "win32" is the value on every Windows build, 64-bit included; there is no "win64".
         if sys.platform == "win32":
-            from logging.handlers import NTEventLogHandler
+            return _WindowsEventLogHandler(SERVICE_NAME)
 
-            return NTEventLogHandler(SERVICE_NAME)
+        if sys.platform.startswith("linux"):
+            from logging.handlers import SysLogHandler
 
-        from logging.handlers import SysLogHandler
+            return SysLogHandler(address="/dev/log", facility=SysLogHandler.LOG_DAEMON)
 
-        return SysLogHandler(address="/dev/log", facility=SysLogHandler.LOG_DAEMON)
+        return logging.NullHandler()
     except Exception:
         # The platform logging handler failed, not `logging` itself; this diagnostic
         # goes to the host application's ordinary logger, never to the audit channel.
@@ -177,8 +257,7 @@ def audit_session_connect(driver_name: str, channel: object, connected: bool) ->
     Channels this package did not create are ignored, so drivers can call this
     unconditionally. A caller who built their own channel never went through
     NI TLS, so there is no transport posture record to pair the outcome with and
-    nothing to attest to; auditing it anyway would also register an Event Log
-    source on machines not using NI TLS at all.
+    nothing to attest to.
     """
     try:
         target = get_channel_target(channel)

tests/unit/test_audit.py

@@ -1,7 +1,11 @@
 "Pytests for nitlsconfig.audit."
 
+import importlib
 import logging
-from typing import Iterator, List
+import logging.handlers
+import sys
+from types import SimpleNamespace
+from typing import Any, Iterator, List, Tuple
 
 import pytest
 
@@ -15,6 +19,7 @@
 
 SERVICE = "ni-grpc-device"
 HOST = "localhost"
+make_logging_handler = audit._make_logging_handler
 
 
 # We utilize a test-specific logging handler to capture audit records for test assertions.
@@ -235,6 +240,203 @@ def test___same_service_requested_twice___reuses_one_logger(
     assert [h for h in logger.handlers if h is recorded] == [recorded]
 
 
+def test___windows_platform___creates_windows_event_log_handler(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    windows_handler = RecordingHandler()
+    monkeypatch.setattr(sys, "platform", "win32")
+    monkeypatch.setattr(audit, "_WindowsEventLogHandler", lambda source: windows_handler)
+
+    assert make_logging_handler() is windows_handler
+
+
+def test___linux_platform___creates_syslog_handler(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    syslog_handler = RecordingHandler()
+    calls: List[Tuple[object, object]] = []
+
+    class FakeSysLogHandler:
+        LOG_DAEMON = 3
+
+        def __new__(cls, address: object, facility: object) -> Any:
+            calls.append((address, facility))
+            return syslog_handler
+
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr(logging.handlers, "SysLogHandler", FakeSysLogHandler)
+
+    assert make_logging_handler() is syslog_handler
+    assert calls == [("/dev/log", FakeSysLogHandler.LOG_DAEMON)]
+
+
+def test___unsupported_platform___disables_audit_logging(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(sys, "platform", "unsupported")
+
+    assert isinstance(make_logging_handler(), logging.NullHandler)
+
+
+@pytest.mark.parametrize(
+    "level, expected_event_type, expected_category",
+    [
+        (logging.INFO, 4, 2),
+        (logging.WARNING, 2, 3),
+        (logging.ERROR, 1, 4),
+    ],
+)
+def test___windows_handler___matches_spdlog_event_type_category_and_message_id(
+    monkeypatch: pytest.MonkeyPatch,
+    level: int,
+    expected_event_type: int,
+    expected_category: int,
+) -> None:
+    # Capture the direct win32evtlog calls in memory. The fake deliberately has no
+    # AddSourceToRegistry API: the source is pre-registered, and this package must
+    # never create or overwrite an Event Log registry key.
+    registered: List[Tuple[object, str]] = []
+    reported: List[Tuple[object, ...]] = []
+    deregistered: List[object] = []
+    event_source = object()
+    user_sid = object()
+
+    class Token:
+        def __init__(self) -> None:
+            self.closed = False
+
+        def Close(self) -> None:  # noqa: N802 - pywin32 API
+            self.closed = True
+
+    token = Token()
+
+    def register_event_source(server: object, source: str) -> object:
+        registered.append((server, source))
+        return event_source
+
+    # Create a fake win32evtlog module with the constants and functions used by the handler.
+    event_log = SimpleNamespace(
+        EVENTLOG_INFORMATION_TYPE=4,
+        EVENTLOG_WARNING_TYPE=2,
+        EVENTLOG_ERROR_TYPE=1,
+        RegisterEventSource=register_event_source,
+        ReportEvent=lambda *args: reported.append(args),
+        DeregisterEventSource=lambda handle: deregistered.append(handle),
+    )
+    modules = {
+        "win32api": SimpleNamespace(GetCurrentProcess=lambda: "process"),
+        "win32con": SimpleNamespace(TOKEN_QUERY=8),
+        "win32evtlog": event_log,
+        "win32security": SimpleNamespace(
+            TokenUser=1,
+            OpenProcessToken=lambda process, access: token,
+            GetTokenInformation=lambda handle, information_class: (user_sid, 0),
+        ),
+    }
+    # Keep the test platform-independent and prevent it from writing a real Windows event.
+    monkeypatch.setattr(importlib, "import_module", modules.__getitem__)
+
+    handler = audit._WindowsEventLogHandler(SERVICE)
+    handler.setFormatter(logging.Formatter("[ni-grpc-device][Client] %(message)s"))
+    record = logging.LogRecord(
+        "nitlsconfig.audit",  # Logger name recorded with the event.
+        level,  # Python level mapped to the equivalent Windows and spdlog values.
+        __file__,  # Source pathname required by LogRecord, but not sent to Event Log.
+        1,  # Source line required by LogRecord, but not sent to Event Log.
+        "session failed",  # Message formatted into the Event Log insertion string.
+        (),  # No %-formatting arguments are needed for this message.
+        None,  # No exception information is associated with this record.
+    )
+
+    handler.emit(record)
+
+    # RegisterEventSource resolves the pre-registered source; None selects the local
+    # Windows machine.
+    assert registered == [(None, SERVICE)]
+    # This source uses mscoree.dll as its EventMessageFile. Its generic event ID 1000
+    # renders the first insertion string as the complete description, avoiding Event
+    # Viewer's "message was not found in the message table" fallback.
+    assert reported == [
+        (
+            event_source,  # Handle returned by RegisterEventSource.
+            expected_event_type,  # Windows information or error classification.
+            expected_category,  # Matches the spdlog level used as its category.
+            1000,  # mscoree.dll event ID 1000 renders the first insertion string literally.
+            user_sid,  # Current process user's security identifier.
+            ["[ni-grpc-device][Client] session failed"],  # Complete event description.
+            None,  # No binary event data.
+        )
+    ]
+    # Every registered source handle must be released after the event is reported.
+    assert deregistered == [event_source]
+    assert token.closed
+
+
+def test___windows_user_sid_unavailable___still_reports_event(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    reported: List[Tuple[object, ...]] = []
+    event_source = object()
+    event_log = SimpleNamespace(
+        EVENTLOG_INFORMATION_TYPE=4,
+        EVENTLOG_WARNING_TYPE=2,
+        EVENTLOG_ERROR_TYPE=1,
+        RegisterEventSource=lambda server, source: event_source,
+        ReportEvent=lambda *args: reported.append(args),
+        DeregisterEventSource=lambda handle: None,
+    )
+
+    def import_module(name: str) -> object:
+        if name == "win32evtlog":
+            return event_log
+        raise OSError("process token is unavailable")
+
+    monkeypatch.setattr(importlib, "import_module", import_module)
+    handler = audit._WindowsEventLogHandler(SERVICE)
+    record = logging.LogRecord("nitlsconfig.audit", logging.INFO, __file__, 1, "message", (), None)
+
+    handler.emit(record)
+
+    assert reported[0][4] is None
+
+
+def test___windows_event_log_failure___is_quiet_and_releases_source(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    event_source = object()
+    deregistered: List[object] = []
+
+    def report_event(*args: object) -> None:
+        # Simulate a transient Windows Event Log failure after the source handle opens.
+        raise OSError("Event Log is unavailable")
+
+    event_log = SimpleNamespace(
+        EVENTLOG_INFORMATION_TYPE=4,
+        EVENTLOG_WARNING_TYPE=2,
+        EVENTLOG_ERROR_TYPE=1,
+        RegisterEventSource=lambda server, source: event_source,
+        ReportEvent=report_event,
+        DeregisterEventSource=lambda handle: deregistered.append(handle),
+    )
+    monkeypatch.setattr(importlib, "import_module", lambda name: event_log)
+    handler = audit._WindowsEventLogHandler(SERVICE)
+    monkeypatch.setattr(audit, "_make_logging_handler", lambda: handler)
+    # Handler.handleError() prints to stderr when this is true. The Windows handler must
+    # instead let the outer audit boundary swallow the failure quietly.
+    monkeypatch.setattr(logging, "raiseExceptions", True)
+    reset_logger()
+
+    audit_transport_posture(HOST, TransportSecurity.MutualTls)
+
+    # The handler's finally block must release a successfully opened source even when
+    # ReportEvent fails.
+    assert deregistered == [event_source]
+    # Regression check: audit sink failures must not leak diagnostics or audit text into
+    # the host application's stderr.
+    assert capsys.readouterr().err == ""
+
+
 def test___logging_handler_cannot_be_created___does_not_raise(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: