logging

logging

File-backed best-effort logger — never raises, never disrupts callers.

BestEffortLogger binds a destination path on construction so any subsystem in any terok-* package can spin up its own log file with one shared, audited idiom.

Writes soft-fail: a logging error must never take down the caller, so every write is wrapped and swallowed. Operator-facing stderr output is run through sanitize_tty so attacker-influenced strings can't smuggle terminal escapes (CWE-150); the file-side write keeps the original bytes for forensic review.

__all__ = ['BestEffortLogger', 'configure'] module-attribute

BestEffortLogger(log_path_fn)

Append timestamped lines to a state-file log; soft-fail on any error.

The destination is supplied as a callable rather than an eager Path so XDG / env-var overrides applied between construction and write time still take effect.

Parameters:

Name	Type	Description	Default
log_path_fn	Callable[[], Path]	Zero-arg callable returning the destination path. Called on every write so tests overriding HOME / XDG_STATE_HOME see their override applied even when the logger was constructed under the previous environment.	required

Bind the destination resolver.

Source code in src/terok_util/logging.py

def __init__(self, log_path_fn: Callable[[], Path]) -> None:
    """Bind the destination resolver."""
    self._log_path_fn = log_path_fn

log(message, *, level='DEBUG')

Append one [timestamp] LEVEL: message line. Never raises.

File creation goes through os.open with mode 0o600 so the log lands owner-only by construction — atomically, without relying on the process umask. The mode bits are honoured by the kernel only on creation; existing files keep whatever perms they were created with.

Source code in src/terok_util/logging.py

def log(self, message: str, *, level: str = "DEBUG") -> None:
    """Append one ``[timestamp] LEVEL: message`` line.  Never raises.

    File creation goes through ``os.open`` with mode ``0o600`` so the
    log lands owner-only by construction — atomically, without
    relying on the process umask.  The mode bits are honoured by
    the kernel only on creation; existing files keep whatever perms
    they were created with.
    """
    try:
        log_path = self._log_path_fn()
        log_path.parent.mkdir(parents=True, exist_ok=True)
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        fd = os.open(
            log_path,
            os.O_APPEND | os.O_CREAT | os.O_WRONLY,
            0o600,
        )
        with os.fdopen(fd, "a", encoding="utf-8") as f:
            f.write(f"[{timestamp}] {level}: {message}\n")
    except Exception:  # nosec B110 — intentionally silent
        pass

debug(message)

Append a DEBUG-level line.

Source code in src/terok_util/logging.py

def debug(self, message: str) -> None:
    """Append a DEBUG-level line."""
    self.log(message, level="DEBUG")

warning(message)

Append a WARNING-level line.

Source code in src/terok_util/logging.py

def warning(self, message: str) -> None:
    """Append a WARNING-level line."""
    self.log(message, level="WARNING")

warn_user(component, message)

Print a structured warning to stderr and append it to the log file.

Stderr output is run through sanitize_tty so attacker bytes in component / message (e.g. originating from foreign config files) can't smuggle terminal escapes into the operator's terminal. The file-side write is unsanitised so the log keeps the original bytes for forensic review.

Source code in src/terok_util/logging.py

def warn_user(self, component: str, message: str) -> None:
    """Print a structured warning to stderr and append it to the log file.

    Stderr output is run through
    [`sanitize_tty`][terok_util.security.sanitize_tty] so attacker
    bytes in *component* / *message* (e.g. originating from foreign
    config files) can't smuggle terminal escapes into the operator's
    terminal.  The file-side write is unsanitised so the log keeps
    the original bytes for forensic review.
    """
    try:
        print(
            f"Warning [{sanitize_tty(component)}]: {sanitize_tty(message)}",
            file=sys.stderr,
        )
    except Exception:  # nosec B110 — intentionally silent
        pass
    self.warning(f"[{component}] {message}")

configure(identifier, *, level=logging.INFO, fmt=_DEFAULT_FORMAT, stream=None, stderr=False)

Install the unified log handler(s) on the root logger — the one call a package makes.

When journald is present the root logger's records go to it (tagged SYSLOG_IDENTIFIER=identifier); otherwise they fall back to a stderr StreamHandler formatted with fmt. Containers have no journal socket, so an in-container daemon always takes the stderr branch — preserving the pattern where a wrapper redirects that stderr to a file.

Set stderr when the process's stderr is deliberately consumed by a parent (a launched daemon whose logs the launcher reads): a stderr handler is then installed in addition to journald, so the parent's pipe keeps receiving records even on a journald host.

Idempotent by construction: handlers previously installed by configure are removed first, so re-invoking it never stacks duplicates. Every module that already does logging.getLogger(__name__) is captured with no call-site change, because all such loggers propagate to the root.

Parameters:

Name	Type	Description	Default
identifier	str	SYSLOG_IDENTIFIER for journald entries and the audit name for the process (e.g. "terok-shield").	required
level	int	Root log level.	INFO
fmt	str	logging.Formatter string for the stderr handler.	_DEFAULT_FORMAT
stream	TextIO | None	Stderr stream to use (defaults to :data:sys.stderr).	None
stderr	bool	Also emit to stderr even when journald is the primary sink — for daemons whose stderr a parent process consumes.	False

Returns:

Type	Description
list[Handler]	The installed handler(s), primary first (for tests / re-wiring).

Source code in src/terok_util/logging.py

def configure(
    identifier: str,
    *,
    level: int = logging.INFO,
    fmt: str = _DEFAULT_FORMAT,
    stream: TextIO | None = None,
    stderr: bool = False,
) -> list[logging.Handler]:
    """Install the unified log handler(s) on the root logger — the one call a package makes.

    When journald is present the root logger's records go to it (tagged
    ``SYSLOG_IDENTIFIER=identifier``); otherwise they fall back to a
    stderr [`StreamHandler`][logging.StreamHandler] formatted with *fmt*.
    Containers have no journal socket, so an in-container daemon always
    takes the stderr branch — preserving the pattern where a wrapper
    redirects that stderr to a file.

    Set *stderr* when the process's stderr is deliberately consumed by a
    parent (a launched daemon whose logs the launcher reads): a stderr
    handler is then installed **in addition** to journald, so the parent's
    pipe keeps receiving records even on a journald host.

    Idempotent by construction: handlers previously installed by
    ``configure`` are removed first, so re-invoking it never stacks
    duplicates.  Every module that already does
    ``logging.getLogger(__name__)`` is captured with no call-site change,
    because all such loggers propagate to the root.

    Args:
        identifier: ``SYSLOG_IDENTIFIER`` for journald entries and the
            audit name for the process (e.g. ``"terok-shield"``).
        level: Root log level.
        fmt: ``logging.Formatter`` string for the stderr handler.
        stream: Stderr stream to use (defaults to :data:`sys.stderr`).
        stderr: Also emit to stderr even when journald is the primary sink
            — for daemons whose stderr a parent process consumes.

    Returns:
        The installed handler(s), primary first (for tests / re-wiring).
    """
    root = logging.getLogger()
    root.setLevel(level)
    for stale in [h for h in root.handlers if getattr(h, _HANDLER_TAG, False)]:
        root.removeHandler(stale)
        stale.close()

    handlers: list[logging.Handler] = []
    if journald_available():
        primary = _JournalHandler(identifier)
        primary.setLevel(level)
        setattr(primary, _HANDLER_TAG, True)
        handlers.append(primary)
        if stderr:
            handlers.append(_stderr_handler(level, fmt, stream))
    else:
        handlers.append(_stderr_handler(level, fmt, stream))

    for handler in handlers:
        root.addHandler(handler)
    return handlers