Data Mining MCP Servers For Vulnerabilities

数据挖掘MCP服务器漏洞

数据挖掘MCP服务器漏洞 介绍/动机 详细的步骤，以便复制

下载MCP服务器

首先，我们需要下载MCP服务器github仓库：

import os
import json
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed

def is_repo_accessible(github_link):
    """Check if a GitHub repository is accessible without authentication."""
    try:
        subprocess.run(
            ["git", "ls-remote", github_link],
            check=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            env={**os.environ, "GIT_TERMINAL_PROMPT": "0"}  # Disable password prompt
        )
        return True
    except subprocess.CalledProcessError:
        return False

def clone_repository(github_link, target_folder):
    """Clone a single repository."""
    repo_name = os.path.basename(github_link.rstrip('/'))
    target_path = os.path.join(target_folder, repo_name)
    try:
        subprocess.run(
            ["git", "clone", github_link, target_path],
            check=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        return True
    except subprocess.CalledProcessError:
        return False

def process_repository(github_link, target_folder):
    """Check if the repository is accessible and clone it if it is."""
    if is_repo_accessible(github_link):
        if clone_repository(github_link, target_folder):
            return "cloned", github_link
        else:
            return "failed", github_link
    else:
        return "inaccessible", github_link

def clone_repositories(json_file, target_folder, max_workers=40):
    """Read repositories from JSON and process them in threads."""
    # Ensure the target folder exists
    os.makedirs(target_folder, exist_ok=True)

    # Read the JSON file
    with open(json_file, 'r') as file:
        data = json.load(file)

    # Lists to track results
    inaccessible_repos = []
    cloned_repos = []
    failed_repos = []

    # Total repositories
    total_repos = len(data)

    # Process repositories in parallel
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(process_repository, repo["github_link"], target_folder): repo["github_link"]
            for repo in data
        }

        # Process results as they complete
        for index, future in enumerate(as_completed(futures), start=1):
            github_link = futures[future]
            try:
                status, repo_link = future.result()
                if status == "cloned":
                    cloned_repos.append(repo_link)
                    print(f"[{index}/{total_repos}] Successfully cloned: {repo_link}")
                elif status == "inaccessible":
                    inaccessible_repos.append(repo_link)
                    print(f"[{index}/{total_repos}] Inaccessible (requires authentication): {repo_link}")
                elif status == "failed":
                    failed_repos.append(repo_link)
                    print(f"[{index}/{total_repos}] Failed to clone: {repo_link}")
            except Exception as e:
                print(f"[{index}/{total_repos}] Error processing {github_link}: {e}")

    # Print results
    print("\nCloning process completed.\n")
    print("Statistics:")
    print(f"Total repositories: {total_repos}")
    print(f"Successfully cloned: {len(cloned_repos)}")
    print(f"Inaccessible repositories: {len(inaccessible_repos)}")
    print(f"Failed to clone: {len(failed_repos)}")

    if inaccessible_repos:
        print("\nInaccessible repositories:")
        for repo in inaccessible_repos:
            print(f"- {repo}")

    if failed_repos:
        print("\nFailed repositories:")
        for repo in failed_repos:
            print(f"- {repo}")

if __name__ == "__main__":
    # Path to the JSON file
    json_file = "github_repos.json"  

    # Path to the folder where repositories will be cloned
    target_folder = "/media/sf_MCP/cloned_repos" 

    # Clone the repositories
    clone_repositories(json_file, target_folder)

在开始寻找漏洞之前，让我们先了解一下我们的数据集。 让我们从一些简单的统计数据开始：

回购总数：3795

跑 总体file_type_count_and_ppercentage.sh

#!/bin/bash

# Directory containing the cloned repositories
base_dir="/media/sf_MCP/cloned_repos"

# Find all files and extract extensions
find "$base_dir" -type f | awk -F. '
  NF>1 {ext[$NF]++; total++} # Increment count for each file extension and total
  END {
    for (e in ext) {
      printf "%.2f %s\n", (ext[e]/total)*100, e # Print percentage first for sorting
    }
  }
' | sort -nr | awk '
  {printf "%s: %.2f%%\n", $2, $1} # Reformat the output
'

输出在https://github.com/BitnomadLive/Data-Mining-MCP-Servers-for-Vulnerabilities/blob/main/Code/statistics/overall_file_type_count_and_precentage_output.txt

跑 count_files_per_repo.sh>count_files_per \_repo_output.txt

#!/bin/bash

# Directory containing the cloned repositories
base_dir="/media/sf_MCP/cloned_repos"

# Temporary file to store the intermediate results
temp_file=$(mktemp)

# Loop through each repository
for repo in "$base_dir"/*; do
  if [ -d "$repo" ]; then
    repo_name=$(basename "$repo")
    file_count=$(find "$repo" -type f | wc -l)
    echo "$file_count $repo_name" >> "$temp_file"
  fi
done

# Sort the results in descending order and display them
sort -nr "$temp_file"

# Clean up the temporary file
rm "$temp_file"

输出在https://raw.githubusercontent.com/BitnomadLive/Data-Mining-MCP-Servers-for-Vulnerabilities/refs/heads/main/Code/statistics/count_files_per_repo_output.txt

生成存储库中文件数量分布的图像

import matplotlib.pyplot as plt
from collections import Counter

def read_file_counts(file_path):
    """
    Reads the file counts and repository names from a text file.
    Args:
        file_path (str): Path to the input file.
    Returns:
        list: A list of file counts (integers).
    """
    file_counts = []
    try:
        with open(file_path, "r") as file:
            for line in file:
                # Split the line into file count and repository name
                parts = line.strip().split(maxsplit=1)
                if len(parts) == 2:
                    file_count = int(parts[0])  # Convert file count to integer
                    file_counts.append(file_count)
    except Exception as e:
        print(f"Error reading file: {e}")
    return file_counts

def plot_file_distribution_bar_chart(file_counts, output_image):
    """
    Plots a bar chart of file counts across repositories.
    Args:
        file_counts (list): A list of file counts (integers).
        output_image (str): Path to save the output image.
    """
    # Count the number of repositories for each unique file count
    count_distribution = Counter(file_counts)

    # Sort the data by file count
    sorted_counts = sorted(count_distribution.items())

    # Extract data for the bar chart
    x_labels = [str(file_count) for file_count, _ in sorted_counts]
    y_values = [repo_count for _, repo_count in sorted_counts]

    # Create the bar chart
    plt.figure(figsize=(30, 10))  # Increase figure size for better readability
    bar_width = 0.8  # Adjust bar width based on number of bars
    plt.bar(x_labels, y_values, color="skyblue", edgecolor="black", width=bar_width)

    # Adjust X-axis labels for readability
    plt.xticks(range(len(x_labels)), x_labels, rotation=45, ha="right", fontsize=6)  # Smaller font size

    # Add labels and title
    plt.xlabel("Number of Files", fontsize=14)
    plt.ylabel("Number of Repositories", fontsize=14)
    plt.title("Distribution of File Counts Across Repositories", fontsize=16)
    plt.tight_layout()

    # Save the image
    plt.savefig(output_image, dpi=300)
    print(f"Bar chart image saved as {output_image}")

    # Show the plot (optional)
    plt.show()

def main():
    """
    Main function to read data and generate the bar chart.
    """
    # Path to the input file
    input_file = "count_files_per_repo_output.txt"
    
    # Output image path
    output_image = "file_distribution_bar_chart_no_bins.png"
    
    # Read the file counts from the file
    file_counts = read_file_counts(input_file)
    
    # Check if data was successfully read
    if file_counts:
        # Call the function to plot the bar chart
        plot_file_distribution_bar_chart(file_counts, output_image)
    else:
        print("No data to plot. Please check the input file.")

# Call the main function
if __name__ == "__main__":
    main()

python 3生成_分发_映像.py

Todo：解释图片

在获得一些统计数据后，看看是否有MCP服务器试图实现类似的事情可能会很有趣。 为了研究这一点，首先将所有README.md文件转换为向量，然后使用DBSCAN算法进行聚类。 DBSCAN（基于密度的带噪声应用程序空间聚类）是一种基于密度对数据点进行分组的聚类算法。它将簇识别为由低密度区域分隔的高点密度区域。该算法需要两个关键参数：eps（被视为邻居的两点之间的最大距离）和minPts（形成密集区域所需的最小点数）。DBSCAN将点分为核心点（密集区域中心）、边界点（在聚类边缘）或噪声点（异常值）。它对于发现任意形状的簇和处理噪声特别有效。 此外，我们将输出集群的数量和每个集群中的repo数量。README文件之间的相似性是通过创建相似性矩阵使用余弦相似性计算的，其中每一行和每一列条目都包含两个README文件间的相似性值。该相似性矩阵被转换为csv文件（output_cytoscape.csv）并写入光盘。此外，所有不适合任何集群的README文件也会写入到一个名为output_with_unclustered.csv的文件中

python3 README_clustering.py

import os
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import DBSCAN
from sklearn.metrics.pairwise import cosine_similarity
from collections import defaultdict

# Function to read README.md files from the root of each repository
def read_readme_files(folder_path):
    repo_readmes = {}
    for repo_name in os.listdir(folder_path):
        repo_path = os.path.join(folder_path, repo_name)
        if os.path.isdir(repo_path):  # Ensure it's a directory
            readme_path = os.path.join(repo_path, "README.md")  # Check for README.md in the root
            if os.path.isfile(readme_path):  # Only add if README.md exists
                with open(readme_path, "r", encoding="utf-8", errors="ignore") as file:
                    repo_readmes[repo_path] = file.read()
    return repo_readmes

# Main script
def main():
    # Folder containing GitHub repositories
    folder_path = "/media/sf_MCP/cloned_repos"  # Updated path

    # Step 1: Read README.md files
    repo_readmes = read_readme_files(folder_path)
    repo_names = list(repo_readmes.keys())
    readme_texts = list(repo_readmes.values())

    # Step 2: Vectorize README.md content
    vectorizer = TfidfVectorizer(stop_words="english", max_features=5000)
    tfidf_matrix = vectorizer.fit_transform(readme_texts)

    # Step 3: Perform DBSCAN clustering
    dbscan = DBSCAN(metric="cosine", eps=0.49, min_samples=2)
    cluster_labels = dbscan.fit_predict(tfidf_matrix)

    # Step 4: Calculate pairwise cosine similarity
    similarity_matrix = cosine_similarity(tfidf_matrix)

    # Step 5: Generate Cytoscape-compatible CSV (no bidirectional duplicates)
    rows = []
    for i in range(len(repo_names)):
        for j in range(i + 1, len(repo_names)):  # Only consider pairs (i, j) where i 导入->文件网络->选择output_cytoscape.csv

1. 将源描述和目标描述更改为源节点属性或目标节点属性
   ![alt text](https://gips3.baidu.com/it/u=2387340077,89898491&fm=3081&app=3081&f=PNG?w=712&h=448 "Cytoscape Import Settings")

1. 单击“确定”导入

1. 布局->边缘加权弹簧嵌入式布局->相似性
   ![alt text](https://gips0.baidu.com/it/u=3040846109,62637280&fm=3081&app=3081&f=PNG?w=978&h=639 "Cytoscape Clusters")
   集群的图片。使用cytoscape.js查看交互式版本

运行分支并提交分析以找出最活跃的仓库：

import os import subprocess import csv from concurrent.futures import ThreadPoolExecutor

Function to check if a directory is a valid Git repository

def is_git_repo(repo_path): try: subprocess.run( ["git", "rev-parse", "--is-inside-work-tree"], cwd=repo_path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True ) return True except subprocess.CalledProcessError: return False

Function to get the number of commits and branches for a repository

def get_repo_stats(repo_path): try: subprocess.run( ["git", "config", "--global", "--add", "safe.directory", repo_path], cwd=repo_path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False ) # Check if the directory is a valid Git repository if not is_git_repo(repo_path): return repo_path, 0, 0

# Get the number of commits commits = subprocess.check_output( ["git", "rev-list", "--count", "HEAD"], cwd=repo_path, stderr=subprocess.DEVNULL, text=True ).strip()

branches = subprocess.check_output( ["git", "--no-pager", "branch", "-r"], cwd=repo_path, stderr=subprocess.DEVNULL, text=True ).splitlines()

# count branches branch_count = len(branches)

return repo_path, int(commits), branch_count except subprocess.CalledProcessError as e: print(f"Error processing repo at {repo_path}: {e}") return repo_path, 0, 0

Function to process all repositories in a folder

def process_repos(folder_path): results = []

# List all subdirectories (potential Git repos) repos = [os.path.join(folder_path, d) for d in os.listdir(folder_path) if os.path.isdir(os.path.join(folder_path, d))]

# Use ThreadPoolExecutor for parallel processing with ThreadPoolExecutor() as executor: futures = [executor.submit(get_repo_stats, repo) for repo in repos] for future in futures: result = future.result() if result: results.append(result)

# Sort results by the number of commits in descending order results.sort(key=lambda x: x[1], reverse=True)

return results

Function to print results in a table format

def print_results(results): print("{:<50} {:<15} {:<15}".format("Repository", "Commits", "Branches")) for repo, commits, branches in results: print(f"{repo:<50} {commits:<15} {branches:<15}")

Function to write results to a CSV file

def write_to_csv(results, output_csv): with open(output_csv, mode="w", newline="") as csvfile: writer = csv.writer(csvfile) writer.writerow(["Repository", "Commits", "Branches"]) writer.writerows(results)

Main function

def main(): # Replace this with your folder path folder_path = "/media/sf_MCP/cloned_repos" # Output CSV file name output_csv = "repo_commit_branch_stats.csv"

# Check if folder exists if not os.path.exists(folder_path): print(f"Error: The folder '{folder_path}' does not exist.") return

print("Processing repositories...") results = process_repos(folder_path)

print("\nResults:") print_results(results)

print(f"\nWriting results to '{output_csv}'...") write_to_csv(results, output_csv) print("Done!")

Entry point of the script

if __name__ == "__main__": main()

|回购|承诺|分支机构|
| --- | --- | --- |
|/media/sf_MCP/cloned_repos/mindsdb|19406|268|\
|/media/sf_MCP/cloned_repos/prisma|11560|446|\
|/media/sf_MCP/cloned_repos/grafbase |7448|9|\
|/media/sf_MCP/cloned_repos/reportix | 1554|18|\
|/media/sf_MCP/cloned_repos/tianji | 1383|14|\
|/media/sf_MCP/cloned_repos/solana试剂盒|1352|6|\
|/media/sf_MCP/cloned_repos/lingo.dev | 1131|25|\
|/media/sf_MCP/cloned_repos/超级内存| 1035|44|\
|/media/sf_MCP/cloned_repos/存储体MCP服务器|925|36|\
|/media/sf_MCP/cloned_repos/MCP本地开发|757|2|\
|/media/sf_MCP/cloned_repos/基本内存|736|5|\
|/media/sf_MCP/cloned_repos/unity MCP|672|5|\
|/media/sf_MCP/cloned_repos/devdb vscode|663|4|\
|/media/sf_MCP/cloned_repos/serena | 611 | 34|\
|/media/sf_MCP/cloned_repos/通量运算符|574|5|\
|/media/sf_MCP/cloned_repos/图集MCP服务器|573|2|\
|/media/sf_MCP/cloned_repos/MCP服务器辅助位桶|530|4|\
|/media/sf_MCP/cloned_repos/genai工具箱|494|50|\
|/media/sf_MCP/cloned_repos/llm-context.py |483|2|\
|/media/sf_MCP/cloned_repos/哨兵MCP|471|79|\
|/media/sf_MCP/cloned_repos/网关|470|14|\
|/media/sf_MCP/cloned_repos/生成器|461|5|\
|/media/sf_MCP/cloned_repos/mongodb镜头|448|4|\
|/media/sf_MCP/cloned_repos/teamretro-MCP服务器| 426 | 13|\
|/media/sf_MCP/cloned_repos/代码库MCP|420|2|\
|/media/sf_MCP/cloned_repos/sourcebot | 406 | 42|

在所有repos上运行trufflehog。多线程不起作用。必须做单线程需要很多时间

import os import subprocess

Define the Trufflehog function

def run_trufflehog(repo_path): """ Run Trufflehog against a given repository. """ try: # Path to the Trufflehog executable trufflehog_path = "/home/vboxuser/Tools/trufflehog" # path to folder containing repos

# Trufflehog flags results_flag = "--results=verified,unknown" json_flag = "--json"

# Build the Trufflehog command command = [ trufflehog_path, "git", f"file://{repo_path}", results_flag, json_flag ]

# Run the command and capture the output print(f"Running Trufflehog on: {repo_path}") result = subprocess.run(command, text=True, capture_output=True)

# Check for errors in the output if result.returncode != 0: print(f"Error running Trufflehog on {repo_path}: {result.stderr}") else: # Save the output to a file output_file = os.path.join(repo_path, "trufflehog_results.json") with open(output_file, "w") as f: f.write(result.stdout) print(f"Trufflehog results saved to: {output_file}")

except Exception as e: print(f"An error occurred while processing {repo_path}: {e}")

Main function

def main(): """ Main function to execute Trufflehog sequentially on a list of repositories, resuming from the last checked repository. """ # Path to the directory containing cloned repositories repos_dir = "/media/sf_MCP/cloned_repos"

# Last successfully checked repository/first repo to check last_checked_repo = "100ms-spl-token-sniper-mcp"

# Get a list of repositories in the directory repos = [os.path.join(repos_dir, repo) for repo in os.listdir(repos_dir) if os.path.isdir(os.path.join(repos_dir, repo))]

# Sort repositories alphabetically to ensure consistent order repos.sort()

# Flag to indicate when to start processing start_processing = False

for repo in repos: # Check if we should skip to the last checked repository if not start_processing: if os.path.basename(repo) == last_checked_repo: start_processing = True # Start processing from this repository else: print(f"Skipping already processed repository: {repo}") continue

# Run Trufflehog on the current repository run_trufflehog(repo)

if __name__ == "__main__": main()

对松露结果进行分析：

import os import json

Path to the folder containing GitHub repositories

base_path = "/media/sf_MCP/cloned_repos"

List to store file paths with at least one "Verified": true result

verified_files = []

Traverse the folder structure

for root, dirs, files in os.walk(base_path): for file in files: if file == "trufflehog_results.json": file_path = os.path.join(root, file)

print("Currently working on: " + str(file_path))

# Skip empty files if os.path.getsize(file_path) == 0: continue

# Read and process the JSON file with open(file_path, "r") as f: try: verified_found = False for line in f: data = json.loads(line) if data.get("Verified") == True: verified_found = True break # No need to check further lines in this file if verified_found: verified_files.append(file_path) except json.JSONDecodeError: print(f"Error decoding JSON in file: {file_path}")

Print the files with at least one "Verified: true" result

print("Files with at least one 'Verified: true' result:") for verified_file in verified_files: print(verified_file)

Print statistics

total_files = len(verified_files) print(f"\nTotal non-empty files with 'Verified: true' results: {total_files}")

结果：
至少有一个“已验证：true”结果的文件：
/media/sf_MCP/cloned_repos/基本MCP服务器/trufflehog_results.json
/media/sf_MCP/cloned_repos/MCP nodejs调试器/trufflehog_results.json
/media/sf_MCP/cloned_repos/MCP服务器neon/trufflehog_results.json
/media/sf_MCP/cloned_repos/mindsdb/trufflehog_results.json

请稍后再次检查。乍一看，它们都不是真正可用的。似乎是postgres和mongodb实例

在所有存储库上运行opengrep：

import os import subprocess import json import shutil from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import signal import sys

Paths

OUTPUT_FOLDER = "opengrep_output" REPO_FOLDER = "/media/sf_MCP/cloned_repos" # Folder containing all pre-downloaded repositories PROGRESS_FILE = "progress.json" OPENGREP_BINARY = "/home/vboxuser/opengrep/opengrep_manylinux_x86" OPENGREP_RULES = "/home/vboxuser/opengrep/opengrep-rules"

Ensure output folder exists

os.makedirs(OUTPUT_FOLDER, exist_ok=True)

Track processed repositories

processed_repos = set()

def load_progress(): """Load progress from the progress file.""" if os.path.exists(PROGRESS_FILE): with open(PROGRESS_FILE, "r") as f: return set(json.load(f)) return set()

def save_progress(): """Save progress to the progress file.""" with open(PROGRESS_FILE, "w") as f: json.dump(list(processed_repos), f)

def signal_handler(sig, frame): """Handle interruption signals (e.g., Ctrl+C).""" print("\nPausing... Saving progress.") save_progress() sys.exit(0)

Register the signal handler

signal.signal(signal.SIGINT, signal_handler)

def run_opengrep_on_batch(batch, batch_index): """Run opengrep on a batch of repositories.""" output_file = os.path.join(OUTPUT_FOLDER, f"batch_{batch_index}.json") print(f"Running opengrep on batch {batch_index}...") subprocess.run([ OPENGREP_BINARY, "scan", f"--sarif-output={output_file}", "-f", OPENGREP_RULES, *batch ], check=True)

def batch_repositories(repos, batch_size): """Yield successive batches of repositories.""" for i in range(0, len(repos), batch_size): yield repos[i:i + batch_size]

Load progress

processed_repos = load_progress()

Get list of all repositories in the local folder

all_repos = [os.path.join(REPO_FOLDER, repo) for repo in os.listdir(REPO_FOLDER) if os.path.isdir(os.path.join(REPO_FOLDER, repo))]

Filter out already processed repositories

remaining_repos = [repo for repo in all_repos if repo not in processed_repos]

BATCH_SIZE = 40

Process remaining repositories in batches

for batch_index, batch in enumerate(batch_repositories(remaining_repos, BATCH_SIZE), start=1): process_batch = False

# Skip batches that are already processed for repo in batch: if repo not in processed_repos: process_batch = True break

if not process_batch: print(f"Skipping already processed batch {batch_index}") continue

try: run_opengrep_on_batch(batch, batch_index) processed_repos.update(batch) save_progress() except subprocess.CalledProcessError as e: print(f"Error running opengrep on batch {batch_index}: {e}")

print("All repositories processed.") if os.path.exists(PROGRESS_FILE): os.remove(PROGRESS_FILE)

运行opengrep结果分析：

import json import os from collections import Counter

Path to the folder containing the result files

folder_path = "output"

Initialize a dictionary to store rule details (name, text, and count)

rule_details = {}

Process all JSON files in the folder

try: for file_name in os.listdir(folder_path): file_path = os.path.join(folder_path, file_name)

# Only process files with a .json extension if file_name.endswith(".json"): try: with open(file_path, "r") as file: data = json.load(file)

# Traverse the JSON structure to count rule names containing "security" runs = data.get("runs", []) for run in runs: results = run.get("results", []) for result in results: rule_id = result.get("ruleId", "Unknown Rule") rule_text = result.get("message", {}).get("text", "No description provided.")

# Only consider rule names containing "security" if "security" in rule_id.lower(): if rule_id not in rule_details: rule_details[rule_id] = {"text": rule_text, "count": 0} rule_details[rule_id]["count"] += 1

except json.JSONDecodeError: print(f"Error: File '{file_name}' is not a valid JSON file. Skipping.") except Exception as e: print(f"Error while processing file '{file_name}': {e}") except FileNotFoundError: print(f"Error: Folder '{folder_path}' not found.") exit(1) except Exception as e: print(f"Error while accessing the folder: {e}") exit(1)

Sort the rules by count in descending order

sorted_rules = sorted(rule_details.items(), key=lambda x: x[1]["count"], reverse=True)

Print the rule names, counts, and their text descriptions

print("Rule Name Counts and Descriptions (Containing 'security', Descending):") for rule, details in sorted_rules: print(f"Rule ID: {rule}") print(f"Count: {details['count']}") print(f"Description: {details['text']}") print("-" * 50)

输出可以在这里找到：https://github.com/BitnomadLive/Data-Mining-MCP-Servers-for-Vulnerabilities/blob/main/Code/opengrep_analysis/opengrep_analysis_output.txt
前10名

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.audit.detect-non-literal-fs-filename Count: 4173 Description: Detected that function argument category has entered the fs module. An attacker could potentially control the location of this file, to include going backwards in the directory with '../'. To address this, ensure that user-controlled variables in file paths are validated.

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.audit.path-traversal.path-join-resolve-traversal Count: 2828 Description: Detected possible user input going into a path.join or path.resolve function. This could possibly lead to a path traversal vulnerability, where the attacker can access arbitrary files stored in the file system. Instead, be sure to sanitize or validate user input first.

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.audit.unsafe-formatstring Count: 2382 Description: Detected string concatenation with a non-literal variable in a util.format / console.log function. If an attacker injects a format specifier in the string, it will forge the log message. Try to use constant values for the format string.

Rule ID: home.vboxuser.opengrep.opengrep-rules.dockerfile.security.missing-user Count: 901 Description: By not specifying a USER, a program in the container may run as 'root'. This is a security hazard. If an attacker can control a process running as root, they may have control over the container. Ensure that the last USER in a Dockerfile is a USER other than 'root'.

Rule ID: home.vboxuser.opengrep.opengrep-rules.python.lang.security.audit.dangerous-subprocess-use-audit Count: 886 Description: Detected subprocess function 'run' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

Rule ID: home.vboxuser.opengrep.opengrep-rules.dockerfile.security.missing-user-entrypoint Count: 577 Description: By not specifying a USER, a program in the container may run as 'root'. This is a security hazard. If an attacker can control a process running as root, they may have control over the container. Ensure that the last USER in a Dockerfile is a USER other than 'root'.

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.html-in-template-string Count: 522 Description: This template literal looks like HTML and has interpolated variables. These variables are not HTML-encoded by default. If the variables contain HTML tags, these may be interpreted by the browser, resulting in cross-site scripting (XSS).

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.audit.detect-non-literal-regexp Count: 293 Description: RegExp() called with a op function argument, this might allow an attacker to cause a Regular Expression Denial-of-Service (ReDoS) within your application as RegExP blocks the main thread. For this reason, it is recommended to use hardcoded regexes instead. If your regex is run on user-controlled input, consider performing input validation or use a regex checking/sanitization library such as https://www.npmjs.com/package/recheck to verify that the regex does not appear vulnerable to ReDoS.

Rule ID: home.vboxuser.opengrep.opengrep-rules.python.sqlalchemy.security.sqlalchemy-execute-raw-query Count: 265 Description: Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.

Rule ID: home.vboxuser.opengrep.opengrep-rules.javascript.lang.security.audit.unsafe-dynamic-method Count: 254 Description: Using non-static data to retrieve and run functions from the object is dangerous. If the data is user-controlled, it may allow executing arbitrary code.