check_data_leakage

Instructions

Implementation Reference

• server.py:570-571 (registration)

  Tool 'check_data_leakage' is registered as an MCP tool via the @mcp.tool() decorator on line 570.

  @mcp.tool()
  def check_data_leakage(

• server.py:571-678 (handler)

  Main handler function for the 'check_data_leakage' tool. It assesses cross-context data exposure risks by evaluating security controls (context isolation, session boundaries, PII detection, output sanitization) and data exposure vectors (cross-user memory, log leakage, third-party sharing). Returns a JSON risk assessment with OWASP A04 reference.

  def check_data_leakage(
      agent_name: str,
      has_context_isolation: bool = False,
      has_session_boundaries: bool = False,
      has_pii_detection: bool = False,
      has_output_sanitization: bool = False,
      shares_memory_across_users: bool = False,
      logs_contain_user_data: bool = False,
      third_party_data_sharing: bool = False,
      caller: str = "",
      api_key: str = "",
  ) -> str:
      """Assess cross-context data exposure risks.
  
      Behavior:
          This tool is read-only and stateless — it produces analysis output
          without modifying any external systems, databases, or files.
          Safe to call repeatedly with identical inputs (idempotent).
          Free tier: 10/day rate limit. Pro tier: unlimited.
          No authentication required for basic usage.
  
      When to use:
          Use this tool when you need structured analysis or classification
          of inputs against established frameworks or standards.
  
      When NOT to use:
          Not suitable for real-time production decision-making without
          human review of results.
  
      Args:
          agent_name (str): The agent name to analyze or process.
          has_context_isolation (bool): The has context isolation to analyze or process.
          has_session_boundaries (bool): The has session boundaries to analyze or process.
          has_pii_detection (bool): The has pii detection to analyze or process.
          has_output_sanitization (bool): The has output sanitization to analyze or process.
          shares_memory_across_users (bool): The shares memory across users to analyze or process.
          logs_contain_user_data (bool): The logs contain user data to analyze or process.
          third_party_data_sharing (bool): The third party data sharing to analyze or process.
          api_key (str): The api key to analyze or process.
  
      Behavioral Transparency:
          - Side Effects: This tool is read-only and produces no side effects. It does not modify
            any external state, databases, or files. All output is computed in-memory and returned
            directly to the caller.
          - Authentication: No authentication required for basic usage. Pro/Enterprise tiers
            require a valid MEOK API key passed via the MEOK_API_KEY environment variable.
          - Rate Limits: Free tier: 10 calls/day. Pro tier: unlimited. Rate limit headers are
            included in responses (X-RateLimit-Remaining, X-RateLimit-Reset).
          - Error Handling: Returns structured error objects with 'error' key on failure.
            Never raises unhandled exceptions. Invalid inputs return descriptive validation errors.
          - Idempotency: Fully idempotent — calling with the same inputs always produces the
            same output. Safe to retry on timeout or transient failure.
          - Data Privacy: No input data is stored, logged, or transmitted to external services.
            All processing happens locally within the MCP server process.
      """
      if err := _check_auth(api_key):
          return err
      if err := _rl(caller):
          return err
  
      issues = []
      if shares_memory_across_users:
          issues.append({"issue": "Memory shared across users (cross-tenant leakage)",
                          "severity": "CRITICAL", "cwe": "CWE-200"})
      if not has_context_isolation:
          issues.append({"issue": "No context isolation between sessions",
                          "severity": "HIGH", "cwe": "CWE-668"})
      if not has_session_boundaries:
          issues.append({"issue": "Session boundaries not enforced",
                          "severity": "HIGH", "cwe": "CWE-488"})
      if not has_pii_detection:
          issues.append({"issue": "No PII/secret detection in agent outputs",
                          "severity": "HIGH", "cwe": "CWE-532"})
      if not has_output_sanitization:
          issues.append({"issue": "Agent outputs not sanitized before delivery",
                          "severity": "MEDIUM", "cwe": "CWE-116"})
      if logs_contain_user_data:
          issues.append({"issue": "Logs contain user data (potential data exposure)",
                          "severity": "MEDIUM", "cwe": "CWE-532"})
      if third_party_data_sharing:
          issues.append({"issue": "Data shared with third parties without explicit controls",
                          "severity": "HIGH", "cwe": "CWE-359"})
  
      risk = "LOW"
      if any(i["severity"] == "CRITICAL" for i in issues):
          risk = "CRITICAL"
      elif any(i["severity"] == "HIGH" for i in issues):
          risk = "HIGH"
      elif issues:
          risk = "MEDIUM"
  
      return json.dumps({
          "agent": agent_name,
          "risk_level": risk,
          "controls": {
              "context_isolation": has_context_isolation,
              "session_boundaries": has_session_boundaries,
              "pii_detection": has_pii_detection,
              "output_sanitization": has_output_sanitization,
          },
          "data_exposure_vectors": {
              "cross_user_memory": shares_memory_across_users,
              "log_leakage": logs_contain_user_data,
              "third_party_sharing": third_party_data_sharing,
          },
          "issues": issues,
          "owasp_ref": "A04 - Data Leakage",
      }, indent=2)