check_prompt_injection

Instructions

Implementation Reference

• server.py:252-342 (handler)

  The main handler function for the 'check_prompt_injection' tool. It takes input_text, caller, and api_key, checks auth and rate limits, scans the input against INJECTION_PATTERNS regex list, checks for special/unicode character encoding attacks, checks for unusually long input, calculates a risk level (SAFE/MEDIUM/HIGH/CRITICAL), and returns a JSON response.

  def check_prompt_injection(
      input_text: str,
      caller: str = "",
      api_key: str = "",
  ) -> str:
      """Check text for prompt injection attack patterns.
  
      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:
          input_text (str): The input text 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
  
      detections = []
      text_lower = input_text.lower()
  
      for i, pattern in enumerate(INJECTION_PATTERNS):
          matches = re.findall(pattern, text_lower, re.IGNORECASE)
          if matches:
              detections.append({
                  "pattern_id": f"INJ-{i+1:03d}",
                  "pattern": pattern,
                  "matches": [str(m) if isinstance(m, str) else str(m) for m in matches[:3]],
                  "severity": "CRITICAL" if i < 3 else "HIGH",
              })
  
      special_chars = sum(1 for c in input_text if ord(c) > 127 or c in '\x00\x01\x02\x03')
      if special_chars > len(input_text) * 0.1 and len(input_text) > 20:
          detections.append({
              "pattern_id": "INJ-SPECIAL",
              "description": "High ratio of special/unicode characters (possible encoding attack)",
              "severity": "MEDIUM",
          })
  
      if len(input_text) > 5000:
          detections.append({
              "pattern_id": "INJ-LENGTH",
              "description": f"Unusually long input ({len(input_text)} chars). May contain hidden instructions.",
              "severity": "LOW",
          })
  
      risk = "SAFE"
      if any(d.get("severity") == "CRITICAL" for d in detections):
          risk = "CRITICAL"
      elif any(d.get("severity") == "HIGH" for d in detections):
          risk = "HIGH"
      elif detections:
          risk = "MEDIUM"
  
      return json.dumps({
          "input_length": len(input_text),
          "risk_level": risk,
          "detections": detections,
          "detection_count": len(detections),
          "recommendation": "Block or sanitize this input before passing to agent."
              if risk in ("CRITICAL", "HIGH") else "Input appears safe.",
          "owasp_ref": "A01 - Prompt Injection",
      }, indent=2)

• server.py:251-252 (registration)

  The @mcp.tool() decorator registers 'check_prompt_injection' as an MCP tool on the FastMCP server instance.

  @mcp.tool()
  def check_prompt_injection(

• server.py:252-256 (schema)

  Input schema for the tool: input_text (required str), caller (optional str with default ''), api_key (optional str with default ''). The return type is str (JSON).

  def check_prompt_injection(
      input_text: str,
      caller: str = "",
      api_key: str = "",
  ) -> str:

• server.py:111-122 (helper)

  INJECTION_PATTERNS — a list of 10 regex patterns used to detect prompt injection attacks (e.g., instruction override, system role impersonation, special tokens, jailbreak phrases, code execution attempts).

  INJECTION_PATTERNS = [
      r"ignore\s+(previous|all|above)\s+(instructions?|prompts?)",
      r"(you\s+are|act\s+as|pretend|roleplay|imagine)\s+.{0,30}(admin|root|system)",
      r"system\s*:\s*",
      r"<\|?(system|im_start|endoftext)\|?>",
      r"\\n\\nHuman:|\\n\\nAssistant:",
      r"IMPORTANT:\s*override",
      r"jailbreak|DAN\s*mode|developer\s*mode",
      r"base64_decode|eval\(|exec\(|__import__",
      r"\{\{.*\}\}",
      r"\\x[0-9a-fA-F]{2}",
  ]

• server.py:31-34 (helper)

  _check_auth helper: validates the API key against the MEOK_API_KEY environment variable.

  def _check_auth(api_key: str = "") -> str | None:
      if _MEOK_API_KEY and api_key != _MEOK_API_KEY:
          return "Invalid API key. Get one at https://meok.ai/api-keys"
      return None

• server.py:42-54 (helper)

  _rl helper: enforces free-tier rate limiting (10 calls/day) for the tool.

  def _rl(caller: str = "anonymous", tier: str = "free") -> Optional[str]:
      if tier == "pro":
          return None
      now = datetime.now()
      cutoff = now - timedelta(days=1)
      _usage[caller] = [t for t in _usage[caller] if t > cutoff]
      if len(_usage[caller]) >= FREE_DAILY_LIMIT:
          return (
              f"Free tier limit ({FREE_DAILY_LIMIT}/day). "
              "Upgrade: https://meok.ai/mcp/owasp-agentic/pro"
          )
      _usage[caller].append(now)
      return None