diff --git a/src/nitlsconfig/audit.py b/src/nitlsconfig/audit.py index a3dca7c..1afd517 100644 --- a/src/nitlsconfig/audit.py +++ b/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,6 +164,9 @@ 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. @@ -95,13 +174,14 @@ def _make_logging_handler() -> logging.Handler: 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) diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py index 3d35932..65c8254 100644 --- a/tests/unit/test_audit.py +++ b/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: