Claude Code transcript - page 1/2

[
  {
    "type": "text",
    "text": "Perfect! Now I have a comprehensive understanding. Let me create a summary document:\n\n## LDAP MCP Server Codebase Structure Summary\n\nBased on my exploration of the LDAP Assistant MCP server codebase, here's a comprehensive overview:\n\n### 1. Main Entry Point\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/main.py`\n\nThis is the primary entry point for the MCP server. It:\n- Defines a `SERVER_REGISTRY` with two supported providers: `dirsrv` (389 Directory Server) and `openldap`\n- Provides the `create_server()` function as the canonical entry point for `fastmcp.json`\n- Allows selection of provider via `LDAP_PROVIDER` environment variable or function parameter\n- Returns an instance of `LDAPAssistantMCP` subclass (either `DirSrvMCP` or `OpenLDAPMCP`)\n\n### 2. MCP Server Configuration\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/fastmcp.json`\n\nDefines how the MCP server is launched:\n- Entry point: `src/main.py:create_server`\n- Transport: stdio\n- Python: 3.13 with `uv` environment manager\n- Dependencies: `fastmcp` (>=2.10.1) and `lib389` (>=3.1.3)\n- Loads server config from `./servers.json` (via `LDAP_SERVERS_CONFIG` env var)\n\n### 3. Base Server Classes (Protocol Communication)\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/ldap_assistant_mcp/server.py`\n\nDefines the base server architecture with:\n\n**LDAPAssistantMCP class** - extends `FastMCP`:\n- Manages multiple server configurations (`server_configs` dict)\n- Provides base connection handling and server management\n- Methods for getting/adding servers and setting defaults\n- CLI override support for connection parameters\n\n**Supporting Data Classes:**\n- `LDAPServerConfig` - Represents connection parameters for a single LDAP server with support for:\n  - Simple authentication\n  - SASL methods (GSSAPI, DIGEST-MD5, EXTERNAL)\n  - Anonymous access\n  - Local server access (requires `is_local=True` and `serverid`)\n  - LDAPI Unix socket connections\n- `LDAPAuthMethod` - Enum for supported auth mechanisms\n- `MCPSettings` - Controls privacy mode (sensitive data redaction) via `expose_sensitive_data` flag\n\n### 4. Provider-Specific Implementations\n\n#### A. 389 Directory Server (DirSrv)\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/dirsrv_mcp/server.py`\n\n**DirSrvMCP class** - Full-featured implementation for 389 Directory Server:\n- Extends `LDAPAssistantMCP`\n- Loads configurations from JSON file (`config_path` parameter or `LDAP_SERVERS_CONFIG` env var)\n- Uses connection manager for multi-server support\n- Privacy/sanitization support through `PrivacySanitizer`\n- Registers prompts for guided workflows (tool navigator, replication diagnostics, performance investigation, health checks, connectivity troubleshooting)\n- Registers resources for accessing cn=config attributes\n\n**Connection Management:**\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/dirsrv_mcp/connection.py`\n\n- `ConnectionManager` - Manages connections to multiple 389 DS instances\n- Creates connections on-demand (no reuse due to lib389 reliability issues)\n- Supports local instances with `lib389.DirSrv.local_simple_allocate()` for access to logs, config files, DSE.ldif\n- Supports LDAPI (Unix socket) connections with SASL EXTERNAL auth\n- Supports remote instances with TCP/LDAP(S)\n\n#### B. OpenLDAP\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/openldap_mcp/server.py`\n\n**OpenLDAPMCP class** - Lightweight implementation for generic OpenLDAP:\n- Minimal tool set (currently just `describe_connection` and `whoami`)\n- Uses standard `ldap` module for connections\n- Supports SIMPLE and SASL_EXTERNAL auth\n\n### 5. Tools Registration and Definition\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/dirsrv_mcp/tools/__init__.py`\n\nCentral registry that imports and exposes all tool registration functions:\n- `register_health_tools` - Health checks, first look, diagnostics\n- `register_user_tools` - User search, listing, detail retrieval\n- `register_group_tools` - Group operations\n- `register_search_tools` - Generic LDAP search (privacy-gated)\n- `register_replication_tools` - Replication status, topology, lag, conflicts\n- `register_performance_tools` - Performance metrics and statistics\n- `register_monitoring_tools` - Monitoring operations\n- `register_index_tools` - Index analysis and configuration\n- `register_config_tools` - Configuration retrieval and comparison\n\n**Tool Registration Pattern** - Each tool module follows this pattern:\n```python\ndef register_<category>_tools(mcp: DirSrvMCP) -> None:\n    @mcp.tool()\n    def tool_name(param1: str, param2: Optional[str] = None) -> Dict[str, Any]:\n        \"\"\"Docstring describes the tool.\"\"\"\n        # Implementation\n```\n\n**Example Tool:** `ldap_search` in `/home/david/projects/codeassociates/ldap-assistant-mcp/src/dirsrv_mcp/tools/search.py`\n- Parameters: `base_dn`, `scope` (BASE/ONELEVEL/SUBTREE), `filter`, `attributes`, `attrs_only`, `limit`, `server_name`\n- Returns structured JSON with search results\n- Privacy-gated: disabled unless `LDAP_MCP_EXPOSE_SENSITIVE_DATA=true`\n\n### 6. Configuration Loading\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/config/loader.py`\n\n**ServerListConfig** - Multi-server configuration container:\n- Loads from JSON file or environment variables\n- Supports fallback to legacy single-server configuration\n- Can be serialized/deserialized to/from JSON\n- Includes MCPSettings for privacy controls\n\n**Key Functions:**\n- `load_config()` - Loads from file or environment (precedence: explicit path > env var > legacy env vars)\n- `initialize_connection_manager()` - Registers servers with a connection manager\n- `save_config()` - Persists configuration to JSON with 0600 permissions\n\n### 7. Privacy & Sanitization\n\n**File:** `/home/david/projects/codeassociates/ldap-assistant-mcp/src/lib/privacy.py`\n\n- `PrivacySanitizer` - Redacts sensitive data (DNs, hostnames, configuration values, replication targets) when `expose_sensitive_data=False`\n- `get_sanitizer()` - Returns singleton instance\n- Each tool module includes sanitization functions that apply redaction to results\n\n### 8. Helper Utilities\n\n**Other files in `src/lib/`:**\n- `datetime_utils.py` - Date/time formatting for LDAP timestamps\n- `ldap_utils.py` - LDAP protocol utilities\n- `value_utils.py` - Value formatting (bytes, floats, integers)\n- `result_formatter.py` - Formats findings with severity levels\n\n### Key Architectural Patterns\n\n1. **FastMCP as Base** - All servers extend `FastMCP` from the `fastmcp` library\n2. **Decorator-based Tool Registration** - Tools use `@mcp.tool()` decorator\n3. **Context Managers for Connections** - `_connection()` context manager in DirSrvMCP handles connection lifecycle\n4. **Privacy by Default** - Sensitive data redacted unless explicitly enabled\n5. **Multi-server Support** - Core design supports managing multiple LDAP instances\n6. **Provider Abstraction** - Server registry pattern allows pluggable implementations\n7. **Local vs Remote** - Distinguishes between local (with file system access) and remote server modes\n\n### MCP Protocol Communication\n\nThe MCP protocol communication is handled by the underlying `fastmcp` library. The custom code:\n- Decorates tool functions with `@self.tool()`\n- Decorates prompt functions with `@self.prompt()`\n- Decorates resource handlers with `@self.resource()`\n- Returns JSON-serializable dictionaries/lists as tool results\n- Leverages FastMCP to handle JSON-RPC protocol, stdio transport, and client communication\n\nAll server classes ultimately call `server.run()` to start the MCP server loop via FastMCP."
  },
  {
    "type": "text",
    "text": "agentId: afa6020 (for resuming to continue this agent's work if needed)"
  }
]

     1→"""Health check tools for 389 Directory Server."""
     2→
     3→from __future__ import annotations
     4→
     5→import copy
     6→import re
     7→from datetime import datetime, timezone
     8→from typing import TYPE_CHECKING, Any, Dict, List, Optional
     9→
    10→from lib389 import lint
    11→from lib389.backend import Backends
    12→from lib389.config import Config, Encryption
    13→from lib389.dirsrv_log import DirsrvAccessLog
    14→from lib389.dseldif import DSEldif, FSChecks
    15→from lib389.monitor import Monitor, MonitorDiskSpace, MonitorLDBM
    16→from lib389.nss_ssl import NssSsl
    17→from lib389.plugins import MemberOfPlugin, ReferentialIntegrityPlugin
    18→from lib389.replica import Replica, Replicas
    19→from lib389.tunables import Tunables
    20→
    21→from src.dirsrv_mcp.connection import is_local_server
    22→from src.lib.result_formatter import Severity, format_finding
    23→from src.lib.value_utils import format_bytes, safe_float, safe_int
    24→
    25→if TYPE_CHECKING:
    26→    from src.dirsrv_mcp.connection import ServerConfig
    27→    from src.dirsrv_mcp.server import DirSrvMCP
    28→
    29→
    30→def _sanitize_health_result(mcp: "DirSrvMCP", result: Dict[str, Any]) -> Dict[str, Any]:
    31→    """Sanitize health check result for privacy mode."""
    32→    if not mcp.privacy_enabled:
    33→        return result
    34→
    35→    sanitizer = mcp.sanitizer
    36→    sanitized = dict(result)
    37→
    38→    # Sanitize server name
    39→    if "server" in sanitized:
    40→        sanitized["server"] = sanitizer.sanitize_server_name(sanitized["server"])
    41→
    42→    # Sanitize server lists
    43→    if "servers_checked" in sanitized and isinstance(sanitized["servers_checked"], list):
    44→        sanitized["servers_checked"] = [
    45→            sanitizer.sanitize_server_name(s) for s in sanitized["servers_checked"]
    46→        ]
    47→    if "servers_failed" in sanitized and isinstance(sanitized["servers_failed"], list):
    48→        sanitized["servers_failed"] = [
    49→            sanitizer.sanitize_server_name(s) for s in sanitized["servers_failed"]
    50→        ]
    51→
    52→    # Sanitize findings
    53→    if "findings" in sanitized and isinstance(sanitized["findings"], list):
    54→        sanitized["findings"] = sanitizer.sanitize_findings(sanitized["findings"])
    55→
    56→    # Sanitize metrics (contains server names, backend names, hostnames)
    57→    if "metrics" in sanitized and isinstance(sanitized["metrics"], dict):
    58→        sanitized["metrics"] = _sanitize_metrics(sanitizer, sanitized["metrics"])
    59→
    60→    if "detailed_metrics" in sanitized and isinstance(sanitized["detailed_metrics"], dict):
    61→        sanitized["detailed_metrics"] = _sanitize_metrics(sanitizer, sanitized["detailed_metrics"])
    62→
    63→    return sanitized
    64→
    65→
    66→def _sanitize_metrics(sanitizer, metrics: Dict[str, Any]) -> Dict[str, Any]:
    67→    """Sanitize metrics dictionary."""
    68→    result = {}
    69→    for server_name, server_data in metrics.items():
    70→        anon_server = sanitizer.sanitize_server_name(server_name)
    71→        if isinstance(server_data, dict):
    72→            result[anon_server] = _sanitize_server_metrics(sanitizer, server_data)
    73→        else:
    74→            result[anon_server] = server_data
    75→    return result
    76→
    77→
    78→def _sanitize_server_metrics(sanitizer, data: Dict[str, Any]) -> Dict[str, Any]:
    79→    """Sanitize server-level metrics."""
    80→    result = {}
    81→    for key, value in data.items():
    82→        if key == "server":
    83→            result[key] = sanitizer.sanitize_server_name(value)
    84→        elif key == "replication" and isinstance(value, dict):
    85→            result[key] = _sanitize_replication_metrics(sanitizer, value)
    86→        elif key == "cache" and isinstance(value, dict):
    87→            result[key] = _sanitize_cache_metrics(sanitizer, value)
    88→        elif key == "disk" and isinstance(value, dict):
    89→            result[key] = _sanitize_disk_metrics(sanitizer, value)
    90→        elif key == "certificates" and isinstance(value, dict):
    91→            result[key] = _sanitize_cert_metrics(value)
    92→        else:
    93→            # Keep numeric metrics and status flags
    94→            result[key] = value
    95→    return result
    96→
    97→
    98→def _sanitize_replication_metrics(sanitizer, repl: Dict[str, Any]) -> Dict[str, Any]:
    99→    """Sanitize replication metrics."""
   100→    result = dict(repl)

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

     1→"""389 Directory Server MCP server implementation."""
     2→
     3→from __future__ import annotations
     4→
     5→import logging
     6→import os
     7→from contextlib import contextmanager
     8→from pathlib import Path
     9→import sys
    10→from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple
    11→
    12→from fastmcp.exceptions import ResourceError, ToolError
    13→from fastmcp.prompts import PromptMessage
    14→from lib389.config import Config
    15→
    16→PROJECT_ROOT = Path(__file__).resolve().parents[2]
    17→if str(PROJECT_ROOT) not in sys.path:
    18→    # Ensure imports like `src.*` work when this file is executed directly.
    19→    sys.path.insert(0, str(PROJECT_ROOT))
    20→
    21→from src.config.loader import load_config
    22→from src.dirsrv_mcp.connection import ConnectionManager, ServerConfig
    23→from src.dirsrv_mcp.tools import (
    24→    register_config_tools,
    25→    register_group_tools,
    26→    register_health_tools,
    27→    register_index_tools,
    28→    register_monitoring_tools,
    29→    register_performance_tools,
    30→    register_replication_tools,
    31→    register_search_tools,
    32→    register_user_tools,
    33→)
    34→from src.ldap_assistant_mcp.server import LDAPAssistantMCP, LDAPServerConfig, MCPSettings
    35→from src.lib.privacy import PrivacySanitizer, get_sanitizer
    36→
    37→__all__ = ["DirSrvMCP"]
    38→
    39→
    40→class DirSrvMCP(LDAPAssistantMCP):
    41→    """FastMCP server exposing 389 Directory Server operations."""
    42→
    43→    def __init__(
    44→        self,
    45→        *,
    46→        config_path: Optional[str] = None,
    47→        servers: Optional[Iterable[LDAPServerConfig]] = None,
    48→        connection_manager: Optional[ConnectionManager] = None,
    49→        settings: Optional[MCPSettings] = None,
    50→        name: str = "389ds-mcp",
    51→        instructions: Optional[str] = None,
    52→        **kwargs: Any,
    53→    ) -> None:
    54→        self.logger = logging.getLogger(self.__class__.__name__)
    55→        merged_servers, loaded_settings = self._collect_servers_and_settings(
    56→            config_path=config_path, extra_servers=servers
    57→        )
    58→
    59→        super().__init__(
    60→            name=name,
    61→            instructions=instructions,
    62→            servers=merged_servers or None,
    63→            **kwargs,
    64→        )
    65→
    66→        # Use provided settings, loaded settings, or defaults
    67→        # Note: We use _mcp_settings to avoid conflict with FastMCP's settings property
    68→        self._mcp_settings = settings or loaded_settings or MCPSettings.from_env()
    69→        self._sanitizer = get_sanitizer()
    70→
    71→        self.connection_manager = connection_manager or ConnectionManager()
    72→        for cfg in self.server_configs.values():
    73→            self.connection_manager.add_server(cfg)
    74→
    75→        self.logger.info(
    76→            "DirSrv MCP initialized with %d server(s): %s (privacy_mode=%s)",
    77→            len(self.server_configs),
    78→            ", ".join(self.server_configs.keys()),
    79→            not self._mcp_settings.expose_sensitive_data,
    80→        )

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

    80→        )
    81→
    82→        self._register_prompts()
    83→        self._register_tools()
    84→        self._register_resources()
    85→
    86→    @property
    87→    def privacy_enabled(self) -> bool:
    88→        """Return True if privacy mode is enabled (sensitive data is redacted)."""
    89→        return not self._mcp_settings.expose_sensitive_data
    90→
    91→    @property
    92→    def sanitizer(self) -> PrivacySanitizer:
    93→        """Return the privacy sanitizer instance."""
    94→        return self._sanitizer
    95→
    96→    # --------------------------------------------------------------------- #
    97→    # Initialization helpers
    98→    # --------------------------------------------------------------------- #
    99→
   100→    def _collect_servers_and_settings(
   101→        self,
   102→        *,
   103→        config_path: Optional[str],
   104→        extra_servers: Optional[Iterable[LDAPServerConfig]],
   105→    ) -> Tuple[List[LDAPServerConfig], Optional[MCPSettings]]:
   106→        """Collect server configurations and settings from config file.
   107→
   108→        Returns:
   109→            Tuple of (server list, settings or None if not loaded from config)
   110→        """
   111→        merged: Dict[str, LDAPServerConfig] = {}
   112→        loaded_settings: Optional[MCPSettings] = None
   113→
   114→        should_load_config = config_path or os.environ.get("LDAP_SERVERS_CONFIG")
   115→        if should_load_config:
   116→            try:
   117→                config = load_config(config_file=config_path)  # type: ignore[arg-type]
   118→            except Exception as exc:
   119→                self.logger.warning("Failed to load multi-server config: %s", exc)
   120→            else:
   121→                for server in config.servers:
   122→                    merged[server.name] = server
   123→                loaded_settings = config.settings
   124→
   125→        if extra_servers:
   126→            for server in extra_servers:
   127→                merged[server.name] = server
   128→
   129→        return list(merged.values()), loaded_settings
   130→
   131→    # --------------------------------------------------------------------- #
   132→    # Registration
   133→    # --------------------------------------------------------------------- #
   134→
   135→    def _register_prompts(self) -> None:
   136→        @self.prompt()
   137→        def tool_navigator(goal: str) -> List[PromptMessage]:
   138→            """Guide users through available tools and their usage."""
   139→
   140→            return [
   141→                PromptMessage(role="user", content=f"Directory task: {goal}"),
   142→                PromptMessage(
   143→                    role="assistant",
   144→                    content=(
   145→                        "Use the available MCP tools to accomplish the task. "
   146→                        "Prefer specialized tools first, falling back to ldap_search for advanced queries.\n\n"
   147→                        "**Health & Diagnostics:**\n"
   148→                        "- first_look: Comprehensive health overview across all servers.\n"
   149→                        "- run_healthcheck: Deep health checks with lint rules.\n\n"
   150→                        "**Performance:**\n"
   151→                        "- get_performance_summary: Combined performance overview.\n"
   152→                        "- get_cache_statistics: Entry/DN/DB cache analysis.\n"
   153→                        "- get_connection_statistics: Connection patterns and FD usage.\n"
   154→                        "- get_operation_statistics: Operation counts by type.\n"
   155→                        "- get_thread_statistics: Thread pool utilization.\n"
   156→                        "- get_resource_utilization: Memory, CPU, disk usage.\n\n"
   157→                        "**Replication:**\n"
   158→                        "- get_replication_status: Comprehensive replica and agreement status.\n"
   159→                        "- get_replication_topology: Map topology across all servers.\n"
   160→                        "- check_replication_lag: Analyze sync status and CSN lag.\n"
   161→                        "- list_replication_conflicts: Find conflict and glue entries.\n"
   162→                        "- get_agreement_status: Detailed agreement information.\n\n"
   163→                        "**User Management:**\n"
   164→                        "- list_active_users / list_locked_users / list_all_users\n"
   165→                        "- search_users_by_name / search_users_by_attribute\n"
   166→                        "- get_user_details\n\n"
   167→                        "**Group Management:**\n"
   168→                        "- list_all_groups\n\n"
   169→                        "**Monitoring:**\n"
   170→                        "- run_monitor\n\n"
   171→                        "**Advanced:**\n"
   172→                        "- ldap_search(base_dn, scope, filter, attributes, attrs_only, limit)\n\n"
   173→                        "State which tool you'll call next and why; keep outputs concise."
   174→                    ),
   175→                ),
   176→            ]
   177→
   178→        @self.prompt()
   179→        def diagnose_replication() -> List[PromptMessage]:

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

   100→    result = dict(repl)
   101→    if "agreements" in result and isinstance(result["agreements"], list):
   102→        result["agreements"] = [
   103→            {
   104→                "name": "[agreement]",
   105→                "consumer": sanitizer.sanitize_hostname(a.get("consumer")),
   106→                "suffix": sanitizer.sanitize_suffix(a.get("suffix")),
   107→            }
   108→            for a in result["agreements"]
   109→        ]
   110→    return result
   111→
   112→
   113→def _sanitize_cache_metrics(sanitizer, cache: Dict[str, Any]) -> Dict[str, Any]:
   114→    """Sanitize cache metrics."""
   115→    result = dict(cache)
   116→    if "backends" in result and isinstance(result["backends"], list):
   117→        result["backends"] = [
   118→            {
   119→                "name": "[backend]",
   120→                "entry_cache_hit_ratio": b.get("entry_cache_hit_ratio"),
   121→                "entry_cache_tries": b.get("entry_cache_tries"),
   122→            }
   123→            for b in result["backends"]
   124→        ]
   125→    return result
   126→
   127→
   128→def _sanitize_disk_metrics(sanitizer, disk: Dict[str, Any]) -> Dict[str, Any]:
   129→    """Sanitize disk metrics."""
   130→    result = dict(disk)
   131→    if "partitions" in result and isinstance(result["partitions"], list):
   132→        result["partitions"] = [
   133→            {
   134→                "partition": "[partition]",
   135→                "usage_percent": p.get("usage_percent"),
   136→                "size": p.get("size"),
   137→                "available": p.get("available"),
   138→            }
   139→            for p in result["partitions"]
   140→        ]
   141→    return result
   142→
   143→
   144→def _sanitize_cert_metrics(certs: Dict[str, Any]) -> Dict[str, Any]:
   145→    """Sanitize certificate metrics."""
   146→    result = dict(certs)
   147→    if "certs" in result and isinstance(result["certs"], list):
   148→        result["certs"] = [
   149→            {
   150→                "subject": "[certificate]",
   151→                "type": c.get("type"),
   152→                "days_until_expiry": c.get("days_until_expiry"),
   153→            }
   154→            for c in result["certs"]
   155→        ]
   156→    return result
   157→
   158→# Check UIDs that ONLY work with local servers (require filesystem or NSS access)
   159→# These are the lint_uid() values returned by the check objects
   160→LOCAL_ONLY_CHECK_UIDS = {
   161→    "fschecks",           # FSChecks - file system permission checks
   162→    "monitor-disk-space", # MonitorDiskSpace - disk space monitoring
   163→    "dseldif",            # DSEldif - DSE.ldif configuration access
   164→    "tls",                # NssSsl - certificate database access (uses "tls" uid)
   165→    "logs",               # DirsrvAccessLog - access log analysis
   166→}
   167→
   168→
   169→# Check objects that can perform lint operations (mirrors lib389 CHECK_OBJECTS)
   170→CHECK_OBJECTS = [
   171→    Config,
   172→    Backends,
   173→    Encryption,
   174→    FSChecks,
   175→    ReferentialIntegrityPlugin,
   176→    MemberOfPlugin,
   177→    MonitorDiskSpace,
   178→    Replica,
   179→    DSEldif,
   180→    NssSsl,
   181→    DirsrvAccessLog,
   182→    Tunables,
   183→]
   184→
   185→
   186→def _get_all_error_codes() -> List[Dict[str, Any]]:
   187→    """Get all known DSLE error codes from lib389.lint module."""
   188→    errors = []
   189→    for name in dir(lint):
   190→        if re.match(r"^DS", name):
   191→            error_def = getattr(lint, name)
   192→            if isinstance(error_def, dict) and "dsle" in error_def:
   193→                errors.append({
   194→                    "code": error_def["dsle"],
   195→                    "severity": error_def.get("severity", "UNKNOWN"),
   196→                    "description": error_def.get("description", ""),
   197→                })
   198→    return sorted(errors, key=lambda x: x["code"])
   199→
   200→
   201→def _list_check_targets(ds) -> Dict[str, Any]:
   202→    """List all check targets and their available lint methods."""
   203→    targets = {}
   204→    for check_class in CHECK_OBJECTS:
   205→        try:
   206→            obj = check_class(ds)
   207→            uid = obj.lint_uid()
   208→            methods = []
   209→            for method_name in dir(obj):
   210→                if method_name.startswith("_lint_"):
   211→                    pretty_name = method_name[6:]  # Remove '_lint_' prefix
   212→                    methods.append(pretty_name)
   213→            if methods:
   214→                targets[uid] = {
   215→                    "object": obj,
   216→                    "methods": sorted(methods),
   217→                }
   218→        except Exception:
   219→            # Some objects may fail to instantiate without proper config
   220→            continue
   221→    return targets
   222→
   223→
   224→def _expand_check_spec(targets: Dict[str, Any], spec: str) -> List[tuple]:
   225→    """Expand a check spec like 'config:*' or 'backends:mappingtree' to list of (uid, method)."""
   226→    checks = []
   227→    if ":" in spec:
   228→        uid_pattern, method_pattern = spec.split(":", 1)
   229→    else:
   230→        uid_pattern = spec
   231→        method_pattern = "*"
   232→
   233→    for uid, target_info in targets.items():
   234→        # Match UID
   235→        if uid_pattern != "*" and uid_pattern != uid:
   236→            continue
   237→
   238→        # Match methods
   239→        for method in target_info["methods"]:
   240→            if method_pattern == "*" or method_pattern == method:
   241→                checks.append((uid, method, target_info["object"]))
   242→
   243→    return checks
   244→
   245→
   246→def _run_single_check(obj, method_name: str) -> List[Dict[str, Any]]:
   247→    """Run a single lint check and return results."""
   248→    results = []
   249→    try:

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

   250→        lint_method = getattr(obj, f"_lint_{method_name}", None)
   251→        if lint_method and callable(lint_method):
   252→            for result in lint_method() or []:
   253→                if isinstance(result, dict):
   254→                    # Add check identifier if not present
   255→                    if "check" not in result:
   256→                        result = copy.deepcopy(result)
   257→                        result["check"] = f"{obj.lint_uid()}:{method_name}"
   258→                    results.append(result)
   259→    except Exception as e:
   260→        # Return error as a finding
   261→        results.append({
   262→            "dsle": "RUNTIME_ERROR",
   263→            "severity": "MEDIUM",
   264→            "description": f"Check failed: {obj.lint_uid()}:{method_name}",
   265→            "items": [],
   266→            "detail": str(e),
   267→            "fix": "Review server logs and verify the server is accessible.",
   268→            "check": f"{obj.lint_uid()}:{method_name}",
   269→        })
   270→    return results
   271→
   272→
   273→def _convert_lib389_result_to_finding(result: Dict[str, Any], server_name: str) -> Dict[str, Any]:
   274→    """Convert a lib389 lint result to the MCP finding format."""
   275→    severity_map = {
   276→        "CRITICAL": Severity.CRITICAL,
   277→        "HIGH": Severity.HIGH,
   278→        "MEDIUM": Severity.MEDIUM,
   279→        "LOW": Severity.LOW,
   280→        "INFO": Severity.INFO,
   281→    }
   282→    # Handle case-insensitive severity matching
   283→    raw_severity = result.get("severity", "MEDIUM").upper()
   284→    severity = severity_map.get(raw_severity, Severity.MEDIUM)
   285→
   286→    items = result.get("items", [])
   287→    items_str = ", ".join(str(item) for item in items) if items else "N/A"
   288→
   289→    return format_finding(
   290→        title=f"[{result.get('dsle', 'UNKNOWN')}] {result.get('description', 'Health check finding')}",
   291→        severity=severity,
   292→        impact=f"Affects: {items_str}",
   293→        details=result.get("detail", "No details available"),
   294→        remediation=result.get("fix", "No remediation steps provided"),
   295→        server=server_name,
   296→        metadata={
   297→            "dsle": result.get("dsle"),
   298→            "check": result.get("check"),
   299→            "items": items,
   300→        },
   301→    )
   302→
   303→
   304→def register_health_tools(mcp: DirSrvMCP) -> None:
   305→    """Register health check tools with the MCP server."""
   306→
   307→    @mcp.tool()
   308→    def first_look() -> Dict[str, Any]:
   309→        """Comprehensive health overview - the go-to tool for "what's wrong with my directory?"
   310→
   311→        Performs a complete health assessment across all configured servers including:
   312→        - Server connectivity and basic health
   313→        - Connection and thread utilization
   314→        - Replication status and errors
   315→        - Cache efficiency (entry cache hit ratios)
   316→        - Disk space usage (local servers only)
   317→        - SSL certificate expiration (local servers only)
   318→
   319→        **Note on local vs remote servers:**
   320→        Most checks work via LDAP and are available for all servers. However, the
   321→        following require local server access (is_local=True with serverid):
   322→        - Disk space monitoring (requires filesystem access)
   323→        - Certificate expiration checking (requires NSS database access)
   324→
   325→        For remote servers, these metrics will show as unavailable in the response.
   326→
   327→        Returns prioritized findings with severity levels and actionable recommendations.
   328→        This should be the first tool called when investigating directory issues.
   329→        """
   330→        server_names = mcp.connection_manager.get_server_names()
   331→
   332→        if not server_names:
   333→            return {
   334→                "type": "first_look",
   335→                "summary": "No servers configured",
   336→                "overall_health": "unknown",
   337→                "critical_count": 1,
   338→                "high_count": 0,
   339→                "medium_count": 0,
   340→                "low_count": 0,
   341→                "info_count": 0,
   342→                "findings": [
   343→                    format_finding(
   344→                        title="No LDAP Servers Configured",
   345→                        severity=Severity.CRITICAL,
   346→                        impact="Cannot perform any directory operations",
   347→                        details=(
   348→                            "The LDAP Assistant has no servers configured. "
   349→                            "Configure at least one server via environment variables "

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

     1→"""User management tools for 389 Directory Server."""
     2→
     3→from __future__ import annotations
     4→
     5→import json
     6→from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional
     7→
     8→from lib389.idm.account import Accounts
     9→from lib389.idm.user import nsUserAccounts
    10→
    11→from src.lib.datetime_utils import convert_datetimes_to_strings
    12→from src.lib.privacy import create_count_only_response, create_privacy_error
    13→
    14→if TYPE_CHECKING:
    15→    from src.dirsrv_mcp.server import DirSrvMCP
    16→
    17→
    18→def register_user_tools(mcp: DirSrvMCP) -> None:
    19→    """Register user management tools with the MCP server."""
    20→
    21→    @mcp.tool()
    22→    def list_all_users(limit: int = 50, server_name: Optional[str] = None) -> Dict[str, Any]:
    23→        """List users in the directory with computed status.
    24→
    25→        Note: In privacy mode (default), returns count only.
    26→        Set LDAP_MCP_EXPOSE_SENSITIVE_DATA=true for full user details.
    27→        """
    28→        target = server_name or mcp.default_server
    29→        with mcp._connection(target) as (name, ds):
    30→            base_dn = mcp._get_base_dn(name)
    31→            users = nsUserAccounts(ds, base_dn)
    32→
    33→            # In privacy mode, return count only
    34→            if mcp.privacy_enabled:
    35→                count = sum(1 for _ in users.list())
    36→                return create_count_only_response("user_list", name, count, mcp.sanitizer)
    37→
    38→            results = _collect_entries(mcp, users.list(), ds, base_dn, limit)
    39→            return {
    40→                "type": "user_list",
    41→                "server": name,
    42→                "total_returned": len(results),
    43→                "limit_applied": limit,
    44→                "items": results,
    45→            }
    46→
    47→    @mcp.tool()
    48→    def search_users_by_name(
    49→        name: str, limit: int = 50, server_name: Optional[str] = None
    50→    ) -> Dict[str, Any]:
    51→        """Search for users by name (uid, cn, givenName, sn, displayName, mail).
    52→
    53→        Note: In privacy mode (default), returns count only.
    54→        Set LDAP_MCP_EXPOSE_SENSITIVE_DATA=true for full user details.
    55→        """
    56→        target = server_name or mcp.default_server
    57→        with mcp._connection(target) as (srv, ds):
    58→            base_dn = mcp._get_base_dn(srv)
    59→            if "*" in name:
    60→                search_filter = (
    61→                    f"(|(uid={name})(cn={name})(givenName={name})(sn={name})"
    62→                    f"(displayName={name})(mail={name}))"
    63→                )
    64→            else:
    65→                search_filter = (
    66→                    f"(|(uid=*{name}*)(cn=*{name}*)(givenName=*{name}*)(sn=*{name}*)"
    67→                    f"(displayName=*{name}*)(mail=*{name}*))"
    68→                )
    69→            users = nsUserAccounts(ds, base_dn)
    70→
    71→            # In privacy mode, return count only
    72→            if mcp.privacy_enabled:
    73→                count = sum(1 for _ in users.filter(search_filter))
    74→                return create_count_only_response("user_search", srv, count, mcp.sanitizer)
    75→
    76→            results = _collect_entries(mcp, users.filter(search_filter), ds, base_dn, limit)
    77→            return {
    78→                "type": "user_search",
    79→                "server": srv,
    80→                "search_term": name,
    81→                "filter_used": search_filter,
    82→                "total_returned": len(results),
    83→                "limit_applied": limit,
    84→                "items": results,
    85→            }
    86→
    87→    @mcp.tool()
    88→    def get_user_details(username: str, server_name: Optional[str] = None) -> Dict[str, Any]:
    89→        """Get detailed information about a specific user.
    90→
    91→        Note: This tool is disabled in privacy mode (default).
    92→        Set LDAP_MCP_EXPOSE_SENSITIVE_DATA=true to enable.
    93→        """
    94→        # Disabled in privacy mode
    95→        if mcp.privacy_enabled:
    96→            return create_privacy_error("get_user_details")
    97→
    98→        target = server_name or mcp.default_server
    99→        with mcp._connection(target) as (srv, ds):
   100→            base_dn = mcp._get_base_dn(srv)
   101→            users = nsUserAccounts(ds, base_dn)
   102→            user = users.get(username)
   103→            record = _build_user_record(mcp, user, ds, base_dn)
   104→            return {"type": "user_details", "server": srv, "username": username, "user": record}
   105→
   106→    @mcp.tool()
   107→    def list_active_users(limit: int = 50, server_name: Optional[str] = None) -> Dict[str, Any]:
   108→        """List active (unlocked) users.
   109→
   110→        Note: In privacy mode (default), returns count only.
   111→        Set LDAP_MCP_EXPOSE_SENSITIVE_DATA=true for full user details.
   112→        """
   113→        target = server_name or mcp.default_server
   114→        with mcp._connection(target) as (srv, ds):
   115→            base_dn = mcp._get_base_dn(srv)
   116→            users = nsUserAccounts(ds, base_dn)
   117→
   118→            # In privacy mode, count active users only
   119→            if mcp.privacy_enabled:
   120→                count = 0

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>

File created successfully at: /home/david/projects/codeassociates/ldap-assistant-mcp/MCP_DEMYSTIFIED.md