diff --git a/paskia/db/lifecycle.py b/paskia/db/lifecycle.py index 103ce2a..06a4b4b 100644 --- a/paskia/db/lifecycle.py +++ b/paskia/db/lifecycle.py @@ -101,7 +101,13 @@ def format_log_uuid( previous: Annotated[dict, "pre"] | None = None, current: Annotated[dict, "post"] | None = None, ) -> Optional[str]: # noqa: UP045 - """Format UUID values/keys/actor labels in transaction logs.""" + """Format UUID values/keys/actor labels and censor secrets in transaction logs.""" + # Censor sensitive OIDC key material regardless of value type, but only + # when formatting the value: path components are passed with the component + # itself as value and must stay visible ("oidc.key = "). + if (path == "oidc.key" or path.endswith(".oidc.key")) and value != "key": + return "" + if not isinstance(value, str): return None diff --git a/paskia/db/logging.py b/paskia/db/logging.py deleted file mode 100644 index a48a54d..0000000 --- a/paskia/db/logging.py +++ /dev/null @@ -1,470 +0,0 @@ -""" -Database change logging with pretty-printed diffs. - -Provides a logger for JSONL database changes that formats diffs -in a human-readable path.notation style with color coding. - -UUIDs are replaced with display names where available, or the full UUID string -for types without display names. -""" - -import logging -import re -import sys -from typing import TYPE_CHECKING, Any -from uuid import UUID - -from kanta.logging import configure_logging as configure_kanta_logging - -if TYPE_CHECKING: - from paskia.db.structs import DB - -logger = logging.getLogger("paskia.db") - -# UUID regex pattern (8-4-4-4-12 hex format) -_UUID_PATTERN = re.compile( - r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" -) - -# Pattern to match control characters and bidirectional overrides -_UNSAFE_CHARS = re.compile( - r"[\x00-\x1f\x7f-\x9f" # C0 and C1 control characters - r"\u200e\u200f" # LRM, RLM - r"\u202a-\u202e" # LRE, RLE, PDF, LRO, RLO - r"\u2066-\u2069" # LRI, RLI, FSI, PDI - r"]" -) - -# ANSI color codes (matching FastAPI logging style) -_RESET = "\033[0m" -_SEP = "\033[38;5;242m" # Dark grey for separators (like host/timing in access log) -_PATH_PREFIX = "\033[38;5;242m" # Dark grey for path prefix (like host in access log) -_PATH_FINAL = "\033[38;5;250m" # Default for final element (like path in access log) -_DELETE = "\033[1;31m" # Red for deletions -_ADD = "\033[0;32m" # Green for additions -_ACTION = "\033[1;34m" # Bold blue for action name -_USER = "\033[0;34m" # Blue for user display - - -def _is_uuid(value: str) -> bool: - """Check if a string is a UUID.""" - return bool(_UUID_PATTERN.match(value)) - - -class UuidResolver: - """Resolve UUIDs to display names or short suffixes. - - Uses the previous state for lookups to show the name before any changes. - """ - - def __init__(self, db: "DB | None" = None, previous: dict | None = None): - self._db = db - self._previous = previous - - def resolve(self, uuid_str: str) -> str: - """Resolve a UUID to its display name or the full UUID string.""" - display = self._get_display_name(uuid_str) - if display: - return display - return uuid_str - - def _get_display_name(self, uuid_str: str) -> str | None: - """Look up display name for a UUID. - - First checks the previous state (to show names before changes), - then falls back to the current database. - """ - # Try previous state first (for showing name before a change) - name = self._lookup_in_previous(uuid_str) - if name: - return name - - # Fall back to current database - return self._lookup_in_db(uuid_str) - - def _lookup_in_previous(self, uuid_str: str) -> str | None: - """Look up display name in the previous state dict.""" - if not self._previous: - return None - - # Check users - if "users" in self._previous and uuid_str in self._previous["users"]: - user_data = self._previous["users"][uuid_str] - if isinstance(user_data, dict) and "display_name" in user_data: - return user_data["display_name"] - - # Check orgs - if "orgs" in self._previous and uuid_str in self._previous["orgs"]: - org_data = self._previous["orgs"][uuid_str] - if isinstance(org_data, dict) and "display_name" in org_data: - return org_data["display_name"] - - # Check roles - if "roles" in self._previous and uuid_str in self._previous["roles"]: - role_data = self._previous["roles"][uuid_str] - if isinstance(role_data, dict) and "display_name" in role_data: - return role_data["display_name"] - - # Check permissions - if ( - "permissions" in self._previous - and uuid_str in self._previous["permissions"] - ): - perm_data = self._previous["permissions"][uuid_str] - if isinstance(perm_data, dict) and "display_name" in perm_data: - return perm_data["display_name"] - - return None - - def _lookup_in_db(self, uuid_str: str) -> str | None: - """Look up display name in the current database.""" - if not self._db: - return None - - try: - uuid_obj = UUID(uuid_str) - except ValueError: - return None - - # Check users - if uuid_obj in self._db.users: - return self._db.users[uuid_obj].display_name - - # Check orgs - if uuid_obj in self._db.orgs: - return self._db.orgs[uuid_obj].display_name - - # Check roles - if uuid_obj in self._db.roles: - return self._db.roles[uuid_obj].display_name - - # Check permissions - if uuid_obj in self._db.permissions: - return self._db.permissions[uuid_obj].display_name - - return None - - -def _format_value( - value: Any, - max_len: int = 60, - resolver: UuidResolver | None = None, -) -> str: - """Format a value for display, truncating if needed. - - If resolver is provided, UUIDs are replaced with display names or short suffixes. - """ - if value is None: - return "null" - - if isinstance(value, bool): - return "true" if value else "false" - - if isinstance(value, (int, float)): - return str(value) - - if isinstance(value, str): - # Check if it's a UUID and resolve to display name - if resolver and _is_uuid(value): - return resolver.resolve(value) - # Filter out control characters and bidirectional overrides - value = _UNSAFE_CHARS.sub("", value) - # Truncate long strings - if len(value) > max_len: - return value[: max_len - 3] + "..." - return value - - if isinstance(value, dict): - if not value: - return "{}" - # Check if all values are True - render as set-like {key1, key2} - all_true = all(v is True for v in value.values()) - parts = [] - for k, v in value.items(): - # Replace UUID keys with display names - key_display = resolver.resolve(k) if resolver and _is_uuid(k) else k - if all_true: - parts.append(key_display) - else: - val_display = _format_value(v, max_len=30, resolver=resolver) - parts.append(f"{key_display}: {val_display}") - return "{" + ", ".join(parts) + "}" - - if isinstance(value, list): - if not value: - return "[]" - parts = [_format_value(v, max_len=30, resolver=resolver) for v in value] - return "[" + ", ".join(parts) + "]" - - # Fallback for other types - text = str(value) - if len(text) > max_len: - text = text[: max_len - 3] + "..." - return text - - -def _format_path(path: list[str], resolver: UuidResolver | None = None) -> str: - """Format a path as dot notation with prefix in dark grey, final in default. - - If resolver is provided, UUIDs in the path are replaced with display names. - """ - if not path: - return "" - - # Replace UUIDs in path with display names - if resolver: - path = [resolver.resolve(p) if _is_uuid(p) else p for p in path] - - if len(path) == 1: - return f"{_PATH_FINAL}{path[0]}{_RESET}" - prefix = ".".join(path[:-1]) - final = path[-1] - return f"{_PATH_PREFIX}{prefix}.{_RESET}{_PATH_FINAL}{final}{_RESET}" - - -def _get_nested(data: dict | None, path: list[str]) -> Any: - """Get a nested value from a dict by path, or None if not found.""" - if data is None: - return None - current = data - for key in path: - if not isinstance(current, dict) or key not in current: - return None - current = current[key] - return current - - -def _collect_changes( - diff: dict, - path: list[str], - changes: list[tuple[str, list[str], Any]], - previous: dict | None, -) -> None: - """ - Recursively collect changes from a diff into a flat list. - - Each change is a tuple of (change_type, path, new_value). - change_type is one of: 'add', 'update', 'delete' - """ - if not isinstance(diff, dict): - # Leaf value - check if it existed before - existed = _get_nested(previous, path) is not None - changes.append(("update" if existed else "add", path, diff)) - return - - for key, value in diff.items(): - if key == "$delete": - # $delete contains a list of keys to delete - if isinstance(value, list): - for deleted_key in value: - changes.append(("delete", path + [str(deleted_key)], None)) - else: - changes.append(("delete", path + [str(value)], None)) - - elif key == "$replace": - # $replace replaces the entire collection at this path - # We need to track what was added and what was deleted - old_collection = _get_nested(previous, path) - old_keys = ( - set(old_collection.keys()) - if isinstance(old_collection, dict) - else set() - ) - new_keys = set(value.keys()) if isinstance(value, dict) else set() - - # Items that existed before but not in new = deleted - for deleted_key in old_keys - new_keys: - changes.append(("delete", path + [str(deleted_key)], None)) - - # Items in new collection - if isinstance(value, dict): - for rkey, rval in value.items(): - existed = rkey in old_keys - changes.append( - ("update" if existed else "add", path + [str(rkey)], rval) - ) - elif value or not old_keys: - # Non-dict replacement or empty replacement with nothing before - changes.append( - ("update" if old_collection is not None else "add", path, value) - ) - - elif key.startswith("$"): - # Other special operations (future-proofing) - changes.append(("add", path, {key: value})) - - else: - # Regular nested key - check if this item existed before - new_path = path + [str(key)] - existed = _get_nested(previous, new_path) is not None - if existed: - # Item exists - recurse to show specific field changes - _collect_changes(value, new_path, changes, previous) - else: - # New item - record as add with full value, don't recurse - changes.append(("add", new_path, value)) - - -def _format_change_lines( - change_type: str, - path: list[str], - value: Any, - resolver: UuidResolver | None = None, -) -> list[str]: - """Format a single change as one or more lines. - - If resolver is provided, UUIDs are replaced with display names. - """ - - # Helper to format a value, checking for censored paths - def fmt_value(v: Any, child_path: list[str]) -> str: - if child_path[-2:] == ["oidc", "key"]: - return f"{_SEP}{_RESET}" - return _format_value(v, resolver=resolver) - - # Helper to format path with UUID replacement - def fmt_path(p: list[str]) -> list[str]: - if resolver: - return [resolver.resolve(x) if _is_uuid(x) else x for x in p] - return p - - formatted_path = fmt_path(path) - - if change_type == "delete": - if len(formatted_path) == 1: - return [f" {_DELETE}{formatted_path[0]} ✗{_RESET}"] - prefix = ".".join(formatted_path[:-1]) - final = formatted_path[-1] - return [f" {_PATH_PREFIX}{prefix}.{_RESET}{_DELETE}{final} ✗{_RESET}"] - - if change_type == "add": - # New item being created - only final element in green - # For dict values, show children on separate indented lines - if isinstance(value, dict) and value: - lines = [] - # First line: path with green final element and grey = - if len(formatted_path) == 1: - lines.append(f" {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET}") - else: - prefix = ".".join(formatted_path[:-1]) - final = formatted_path[-1] - lines.append( - f" {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}" - ) - # Child lines: indented key: value, with aligned values - # Format keys (may contain UUIDs) - formatted_items = [] - for k, v in value.items(): - k_display = resolver.resolve(k) if resolver and _is_uuid(k) else k - v_str = fmt_value(v, path + [k]) - formatted_items.append((k_display, v_str)) - max_key_len = max(len(k) for k, _ in formatted_items) - field_width = max(max_key_len, 12) # minimum 12 chars - for k_display, v_str in formatted_items: - padding = " " * (field_width - len(k_display)) - lines.append(f" {k_display}{_SEP}:{_RESET}{padding} {v_str}") - return lines - else: - value_str = fmt_value(value, path) - if len(formatted_path) == 1: - return [ - f" {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET} {value_str}" - ] - prefix = ".".join(formatted_path[:-1]) - final = formatted_path[-1] - return [ - f" {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET} {value_str}" - ] - - # update: Existing item being updated - normal path colors - value_str = fmt_value(value, path) - path_str = _format_path(path, resolver=resolver) - return [f" {path_str} {_SEP}={_RESET} {value_str}"] - - -def format_diff( - diff: dict, previous: dict | None = None, db: "DB | None" = None -) -> list[str]: - """ - Format a JSON diff as human-readable lines. - - Args: - diff: The JSON diff dict - previous: The previous state dict (for determining add vs update) - db: Optional database for looking up display names - - Returns a list of formatted lines (without newlines). - UUIDs are replaced with display names (using previous state for lookups). - """ - changes: list[tuple[str, list[str], Any]] = [] - _collect_changes(diff, [], changes, previous) - - if not changes: - return [] - - # Create resolver for UUID replacement (uses previous state for lookups) - resolver = UuidResolver(db, previous) - - # Format each change - lines = [] - for change_type, path, value in changes: - lines.extend(_format_change_lines(change_type, path, value, resolver)) - - return lines - - -def format_action_header(action: str, user_display: str | None = None) -> str: - """Format the action header line.""" - action_str = f"{_ACTION}{action}{_RESET}" - if user_display: - user_str = f"{_USER}{user_display}{_RESET}" - return f"{action_str} by {user_str}" - return action_str - - -def log_change( - action: str, - diff: dict, - user_display: str | None = None, - previous: dict | None = None, - db: "DB | None" = None, -) -> None: - """ - Log a database change with pretty-printed diff. - - UUIDs are replaced with display names for readability. For types without - display names, the full UUID string is used. - - Args: - action: The action name (e.g., "login", "admin:delete_user") - diff: The JSON diff dict - user_display: Optional display name of the user who performed the action - previous: The previous state dict (for determining add vs update) - db: Optional database for looking up display names - """ - header = format_action_header(action, user_display) - diff_lines = format_diff(diff, previous, db) - - if not diff_lines: - logger.info(header) - return - - if len(diff_lines) == 1: - # Single change - combine on one line - logger.info(f"{header}{diff_lines[0]}") - else: - # Multiple changes - header on its own line, then changes - logger.info(header) - for line in diff_lines: - logger.info(line) - - -def configure_db_logging() -> None: - """Configure the database logger to output to stderr without prefix.""" - handler = logging.StreamHandler(sys.stderr) - handler.setFormatter(logging.Formatter("%(message)s")) - logger.addHandler(handler) - logger.setLevel(logging.INFO) - logger.propagate = False - # Kanta logs changes through its own logger; wire it to the same output. - configure_kanta_logging() diff --git a/paskia/fastapi/mainapp.py b/paskia/fastapi/mainapp.py index 3c4e45c..d502c8f 100644 --- a/paskia/fastapi/mainapp.py +++ b/paskia/fastapi/mainapp.py @@ -7,12 +7,12 @@ from pathlib import Path import msgspec from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, RedirectResponse +from kanta.logging import configure_logging as configure_kanta_logging from paskia import authcode, db, remoteauth from paskia.bootstrap import bootstrap_if_needed from paskia.db.background import start_background, stop_background from paskia.db.lifecycle import kanta -from paskia.db.logging import configure_db_logging from paskia.fastapi import admin, api, auth_host, oid, ws from paskia.fastapi.admin.adminapp import adminapp @@ -26,7 +26,7 @@ from paskia.util.runtime import RuntimeConfig # Configure custom logging configure_access_logging() -configure_db_logging() +configure_kanta_logging() _access_logger = logging.getLogger("paskia.access") diff --git a/pyproject.toml b/pyproject.toml index ea92a45..536dd2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ dependencies = [ "msgspec>=0.20.0", "fastapi-vue>=1.1.0", "ua-parser[regex]>=1.0.1", - "kanta>=0.4.0", + "kanta>=0.7.0", ] [dependency-groups] dev = [