interactsh_stop

Instructions

Implementation Reference

• src/mcp_security_toolkit/tools/interactsh.py:299-352 (handler)

  The core handler for the interactsh_stop tool. Validates token, reads the session file, sends SIGTERM to the process group, optionally deletes the log file (with path traversal protection), removes the session descriptor, and returns a status dict.

  def interactsh_stop(token: str, delete_log: bool = True) -> dict:
      """Stop a previously-registered interactsh-client session and clean up.
  
      Terminates the spawned `interactsh-client` process (best-effort) and
      removes the session descriptor. The log file is removed by default
      (set `delete_log=False` to keep it for post-mortem).
  
      Args:
          token: token returned by `interactsh_register`.
          delete_log: also remove the session log file (default True).
  
      Returns:
          {"stopped": bool, "log_removed": bool, "note": str | None}
      """
      if not _validate_token(token):
          return {"error": "invalid token format (expected 12 hex chars from interactsh_register)"}
  
      sp = _session_path(token)
      if not sp.exists():
          return {"error": f"unknown token: {token}"}
  
      try:
          session = json.loads(sp.read_text())
      except (OSError, json.JSONDecodeError) as e:
          return {"error": f"failed to read session: {e}"}
  
      stopped = False
      pid = session.get("pid")
      if isinstance(pid, int) and pid > 0:
          try:
              os.killpg(os.getpgid(pid), signal.SIGTERM)
              stopped = True
          except (ProcessLookupError, PermissionError, OSError):
              stopped = False
  
      log_removed = False
      if delete_log:
          log = Path(session.get("log_path", ""))
          # Defense in depth: never unlink anything outside our sessions dir,
          # even if a tampered session descriptor points elsewhere.
          if log.exists() and _path_is_inside(log, SESSIONS_DIR):
              try:
                  log.unlink()
                  log_removed = True
              except OSError:
                  log_removed = False
  
      sp.unlink(missing_ok=True)
  
      return {
          "stopped": stopped,
          "log_removed": log_removed,
          "note": None if stopped else "process already exited or no permission to signal",
      }

• src/mcp_security_toolkit/server.py:36-36 (registration)

  Registers interactsh_stop as an MCP tool on the FastMCP server.

  mcp.tool()(interactsh.interactsh_stop)

• src/mcp_security_toolkit/tools/interactsh.py:51-52 (helper)

  Validates token format (12 hex characters) before stop processing.

  def _validate_token(token: str) -> bool:
      return isinstance(token, str) and bool(TOKEN_REGEX.match(token))

• src/mcp_security_toolkit/tools/interactsh.py:94-100 (helper)

  Returns the filesystem path for a session descriptor JSON file.

  def _session_path(token: str) -> Path:
      SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
      try:
          os.chmod(SESSIONS_DIR, 0o700)
      except OSError:
          pass
      return SESSIONS_DIR / f"{token}.json"

• src/mcp_security_toolkit/tools/interactsh.py:59-65 (helper)

  Defense-in-depth check ensuring log deletion doesn't escape the sessions directory.

  def _path_is_inside(child: Path, parent: Path) -> bool:
      """True iff resolved `child` is at or below resolved `parent`."""
      try:
          child.resolve().relative_to(parent.resolve())
          return True
      except (ValueError, OSError):
          return False