ai generated analyze_plugins (i swear i checked it tho)

Reviewed-on: #11

WordPress-Metrics/README.md

@@ -0,0 +1,18 @@
+# WordPress Metrics
+
+## scrape_wordpress_plugins.py
+- Uses requests to scrape the whole plugins list using https://api.wordpress.org/plugins/
+- Scrape in pages of 100 plugins, takes about ~30 mins to complete
+- Stores as output.json 
+
+## analyze_data.py
+- Calculate basic statistics (percentiles, mean, median, sd ...)
+- Visualize them with matplot in a logrithrimic scale
+
+## query_plugin.py
+- Change slug to equal the target plugins (should be able to directly plug the outputs from detect_plugins.sh)
+- TODO: Implement warning system
+
+## security_rating.py
+- Implements some sort of basic rating system factoring in number of active installations, last-update date, and ratings
+- sample use case: python3 security_rating.py {plugin name}

WordPress-Metrics/analyze_data.py

@@ -0,0 +1,74 @@
+import json
+import matplotlib.pyplot as plt
+import numpy as np
+
+def print_info(name, data):
+    # Convert the list to a NumPy array
+    data_np = np.array(data)
+
+    # Calculate basic statistics
+    mean = np.mean(data_np)
+    median = np.median(data_np)
+    std_dev = np.std(data_np)
+    variance = np.var(data_np)
+    min_val = np.min(data_np)
+    max_val = np.max(data_np)
+    sum_val = np.sum(data_np)
+
+    # Calculate percentiles
+    percentile_25 = np.percentile(data_np, 25)
+    percentile_50 = np.percentile(data_np, 50)  # This is the same as the median
+    percentile_75 = np.percentile(data_np, 75)
+    percentile_90 = np.percentile(data_np, 90)
+
+    # Print the results
+    print(f"\n\nDescriptive Statistics for {name}:")
+    print(f"Mean: {mean}")
+    print(f"Median: {median}")
+    print(f"Standard Deviation: {std_dev}")
+    print(f"Variance: {variance}")
+    print(f"Minimum: {min_val}")
+    print(f"Maximum: {max_val}")
+    print(f"Sum: {sum_val}")
+    print("\nPercentiles:")
+    print(f"25th Percentile: {percentile_25}")
+    print(f"50th Percentile (Median): {percentile_50}")
+    print(f"75th Percentile: {percentile_75}")
+    print(f"90th Percentile: {percentile_90}")
+
+
+with open('output.json', 'r') as f:
+    json_data = json.load(f)
+
+# print(len(data))
+downloaded = [plugin["downloaded"] for plugin in json_data]
+rating = [plugin["rating"] for plugin in json_data]
+active_installs = [plugin["active_installs"] for plugin in json_data]
+num_ratings = [plugin["num_ratings"] for plugin in json_data]
+
+print_info("downloaded", downloaded)
+print_info("rating", rating)
+print_info("active_installs", active_installs)
+print_info("num_ratings", num_ratings)
+
+# plt.hist(downloaded, bins=500, log=True)
+# plt.show()
+fig, axs = plt.subplots(nrows=2, ncols=2, figsize=(8, 6))
+axs[0,0].hist(downloaded, bins=100, log=True)
+axs[0,0].set_title("Distribution of downloaded")
+
+axs[0,1].hist(rating, bins=20, log=True)
+axs[0,1].set_title("Distribution of rating")
+
+axs[1,0].hist(active_installs, bins=10, log=True)
+axs[1,0].set_title("Distribution of active_installs")
+
+axs[1,1].hist(num_ratings, bins=100, log=True)
+axs[1,1].set_title("Distribution of num_ratings")
+
+# Add labels and title
+# plt.xlabel("Value")
+# plt.ylabel("Frequency (log)")
+
+# Display the plot
+plt.show()

WordPress-Metrics/analyze_plugins.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Analyze WordPress plugins for security vulnerabilities using security ratings."""
+import argparse
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Import security_rating functions
+from security_rating import get_plugin_info, calculate_security_score
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments.
+    
+    Returns:
+        Parsed arguments namespace.
+    """
+    parser = argparse.ArgumentParser(
+        description="Analyze WordPress plugins for security issues using security ratings."
+    )
+    parser.add_argument(
+        "--input",
+        "-i",
+        required=True,
+        help="Path to input JSON file (from output_recursive_plugins)"
+    )
+    parser.add_argument(
+        "--output",
+        "-o",
+        help="Path to output JSON file (if not specified, prints to stdout)",
+        default=None
+    )
+    parser.add_argument(
+        "--detector-name",
+        "-d",
+        help="Name of detector to use in output (default: Security Rating Analyzer)",
+        default="Security Rating Analyzer"
+    )
+    return parser.parse_args()
+
+
+def analyze_plugin(slug: str) -> dict:
+    """Analyze a single plugin and return its status.
+    
+    Args:
+        slug: Plugin slug to analyze.
+        
+    Returns:
+        Dict with status and reason fields.
+    """
+    plugin_info = get_plugin_info(slug)
+    
+    if not plugin_info:
+        return {
+            "status": "unknown",
+            "reason": f"Could not fetch plugin information from WordPress.org API"
+        }
+    
+    score = calculate_security_score(plugin_info)
+    
+    # Determine status based on score
+    # Safe: score >= 65 (Excellent or Good)
+    # Vulnerable: score < 65 (Moderate or Poor)
+    if score >= 85:
+        status = "safe"
+        reason = f"Excellent security rating ({score}/100) — Actively maintained and trusted"
+    elif score >= 65:
+        status = "safe"
+        reason = f"Good security rating ({score}/100) — Generally safe but monitor updates"
+    elif score >= 45:
+        status = "vulnerable"
+        reason = f"Moderate security rating ({score}/100) — May be outdated or less popular"
+    else:
+        status = "vulnerable"
+        reason = f"Poor security rating ({score}/100) — High risk of security issues"
+    
+    result = {
+        "status": status,
+        "reason": reason,
+        "score": score
+    }
+    
+    # Add version if available
+    version = plugin_info.get("version")
+    if version:
+        result["version"] = version
+    
+    return result
+
+
+def analyze_directory(directory_name: str, plugins_dict: dict, detector_name: str) -> dict:
+    """Analyze all plugins in a directory.
+    
+    Args:
+        directory_name: Name of the directory being analyzed.
+        plugins_dict: Dict mapping plugin slug to absolute path.
+        detector_name: Name of the detector.
+        
+    Returns:
+        Analysis result in the specified format.
+    """
+    timestamp = datetime.now(timezone.utc).isoformat()
+    
+    analyzed_plugins = {}
+    
+    print(f"\nAnalyzing plugins for directory: {directory_name}")
+    print(f"Found {len(plugins_dict)} plugins to analyze")
+    
+    for slug, path in plugins_dict.items():
+        print(f"  Analyzing {slug}...", end=" ", flush=True)
+        result = analyze_plugin(slug)
+        analyzed_plugins[slug] = result
+        print(f"{result['status']}")
+    
+    return {
+        "directory": directory_name,
+        "wordpress": True,
+        "name_of_detector": detector_name,
+        "timestamp": timestamp,
+        "plugins": analyzed_plugins
+    }
+
+
+def main():
+    """Main entry point for the plugin analyzer."""
+    args = parse_args()
+    
+    # Load input JSON
+    try:
+        with open(args.input, "r", encoding="utf-8") as f:
+            input_data = json.load(f)
+    except FileNotFoundError:
+        print(f"Error: Input file '{args.input}' not found.", file=sys.stderr)
+        sys.exit(1)
+    except json.JSONDecodeError as e:
+        print(f"Error: Invalid JSON in '{args.input}': {e}", file=sys.stderr)
+        sys.exit(1)
+    
+    # Analyze each directory
+    results = []
+    
+    for directory_name, plugins_dict in input_data.items():
+        if not plugins_dict:
+            print(f"Skipping {directory_name} (no plugins found)")
+            continue
+        
+        analysis = analyze_directory(directory_name, plugins_dict, args.detector_name)
+        results.append(analysis)
+    
+    # Output results
+    if args.output:
+        with open(args.output, "w", encoding="utf-8") as f:
+            json.dump(results, f, ensure_ascii=False, indent=2)
+        print(f"\nWrote analysis results to {args.output}")
+    else:
+        print("\nAnalysis results:")
+        print(json.dumps(results, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

WordPress-Metrics/query_plugin.py

@@ -0,0 +1,14 @@
+import requests
+
+slug = "akismet"
+response = requests.get("https://api.wordpress.org/plugins/info/1.2/?action=query_plugins"
+                        f"&slug={slug}")
+response_json = response.json()
+response_json = response_json['plugins'][0]
+# print(response_json)
+print(response_json['downloaded'])
+print(response_json['rating'])
+print(response_json['active_installs'])
+print(response_json['num_ratings'])
+
+# TODO: Implement warning system

WordPress-Metrics/scrape_wordpress_plugins.py

@@ -0,0 +1,37 @@
+from time import sleep
+
+import requests
+import json
+
+from tqdm import tqdm
+
+starting_page = 1
+request_per_page = 100
+response = requests.get(f"https://api.wordpress.org/plugins/info/1.2/?action=query_plugins"
+                        f"&request[page]={starting_page}"
+                        f"&request[per_page]={request_per_page}")
+response_json = response.json()
+# print(response_json)
+
+page_no = response_json['info']['pages']
+plugins_info_list = []
+plugins_info_str = ""
+for i in tqdm(range(page_no)):
+    response = requests.get(f"https://api.wordpress.org/plugins/info/1.2/?action=query_plugins"
+                            f"&request[page]={i+1}"
+                            f"&request[per_page]={request_per_page}")
+    response_json = response.json()
+    plugins_info_list.append(response_json['plugins'])
+    # plugins_info_str += str(response_json['plugins'])
+    sleep(0.1)
+
+flattened_list = [item for sublist in plugins_info_list for item in sublist]
+
+plugins_info_json = json.dumps(flattened_list, indent=4)
+print(plugins_info_json)
+
+filename = "output.json"
+
+# Open the file in write mode ('w') and use json.dump() to write the data
+with open(filename, 'w') as f:
+    f.write(plugins_info_json)

WordPress-Metrics/security_rating.py

@@ -0,0 +1,101 @@
+import requests
+import datetime
+import sys
+
+from dateutil import parser
+import datetime
+
+def get_plugin_info(slug):
+    """Fetch plugin info from the WordPress API."""
+    url = f"https://api.wordpress.org/plugins/info/1.2/?action=plugin_information&request[slug]={slug}"
+    response = requests.get(url)
+    if response.status_code != 200:
+        print(f"Error: Unable to fetch data for plugin '{slug}'.")
+        return None
+    return response.json()
+
+def calculate_security_score(plugin):
+    """Calculate a security score (0-100) for a WordPress plugin."""
+    score = 0
+
+    # last updated
+    last_updated_str = plugin.get("last_updated")
+    if last_updated_str:
+        try:
+            last_updated = parser.parse(last_updated_str)
+            now = datetime.datetime.now(datetime.timezone.utc)
+            if last_updated.tzinfo is None:
+                last_updated = last_updated.replace(tzinfo=datetime.timezone.utc)
+            days_since_update = (now - last_updated).days
+
+            if days_since_update <= 90:
+                score += 40
+            elif days_since_update <= 180:
+                score += 30
+            elif days_since_update <= 365:
+                score += 20
+            else:
+                score += 10
+        except Exception as e:
+            print(f"Warning: Could not parse date '{last_updated_str}': {e}")
+            score += 0
+    else:
+        score += 0
+
+    # user rating
+    rating = plugin.get("rating")
+    if isinstance(rating, (int, float)):
+        score += (rating / 100) * 30
+    else:
+        score += 0
+
+    # active installs
+    installs = plugin.get("active_installs")
+    if isinstance(installs, int):
+        if installs >= 1_000_000:
+            score += 20
+        elif installs >= 100_000:
+            score += 15
+        elif installs >= 10_000:
+            score += 10
+        elif installs > 0:
+            score += 5
+    else:
+        score += 0
+
+    return round(score, 1)
+
+
+def rate_plugin(slug):
+    plugin = get_plugin_info(slug)
+    if not plugin:
+        return
+
+    score = calculate_security_score(plugin)
+    name = plugin.get("name", slug)
+    last_updated = plugin.get("last_updated", "Unknown")
+    rating = plugin.get("rating", "Unknown")
+    installs = plugin.get("active_installs", "Unknown")
+
+    print(f"\n🔍 Security Rating for '{name}'")
+    print("=" * (25 + len(name)))
+    print(f"Last Updated: {last_updated}")
+    print(f"User Rating: {rating}/100")
+    print(f"Active Installs: {installs}")
+    print(f"→ Security Score: {score}/100")
+
+    if score >= 85:
+        print("Excellent — Actively maintained and trusted")
+    elif score >= 65:
+        print("Good — Generally safe but monitor updates")
+    elif score >= 45:
+        print("Moderate — May be outdated or less popular")
+    else:
+        print("Poor — High risk of security issues")
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python plugin_security_rating.py <plugin-slug>")
+        sys.exit(1)
+    slug = sys.argv[1]
+    rate_plugin(slug)

detect_plugins.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+# Usage:
+#   ./export-wp-plugins.sh [base_directory] [output_file]
+
+BASE_DIR="${1:-/var/www}"      
+OUTPUT_FILE="${2:-plugins.csv}"
+
+# Check for WP-CLI
+if ! command -v wp &>/dev/null; then
+  echo "WP-CLI not found. Install WP-CLI first: https://wp-cli.org/"
+  exit 1
+fi
+
+# Loop through subdirectories
+for dir in "$BASE_DIR"/*/; do
+  if [ -f "$dir/wp-config.php" ]; then
+    site=$(basename "$dir")
+    echo "=== $site ===" | tee -a "$OUTPUT_FILE"
+    wp --allow-root --path="$dir" plugin list --format=csv>> "$OUTPUT_FILE"
+    echo "" >> "$OUTPUT_FILE"
+  fi
+done
+
+echo "Plugins saved to $OUTPUT_FILE"

old-scripts/CVSS_scores.py

@@ -1,40 +0,0 @@
-import json
-import matplotlib.pyplot as plt
-
-
-json_path = "wordfence_vulnerabilities.json"
-
-def load_cvss_scores(path):
-    """Load JSON data and extract all CVSS scores into a list"""
-    with open(path, "r", encoding="utf-8") as f:
-        data = json.load(f)
-
-    scores = []
-    for entry in data:
-      
-        if "cvss_score" in entry and isinstance(entry["cvss_score"], (int, float)):
-            scores.append(entry["cvss_score"])
-    return scores
-
-
-def plot_hist(scores):
-    """Plot a histogram of CVSS scores"""
-    plt.figure(figsize=(8, 5))
-    plt.hist(scores, bins=10, edgecolor="black", color="skyblue")
-    plt.title("Distribution of CVSS Scores in Vulnerability Data")
-    plt.xlabel("CVSS Score")
-    plt.ylabel("Number of Vulnerabilities")
-    plt.grid(alpha=0.4)
-    plt.savefig("cvss_distribution.png")
-    plt.show()
-
-
-if __name__ == "__main__":
-    scores = load_cvss_scores(json_path)
-    print(f"Loaded {len(scores)} CVSS scores.")
-    print("Sample:", scores[:10])
-
-    if scores:
-        plot_hist(scores)
-    else:
-        print("error")

old-scripts/top_vulnerability_types.py

@@ -1,49 +0,0 @@
-#!/usr/bin/env python3
-import json
-from collections import Counter
-import matplotlib.pyplot as plt
-
-
-path = "wordfence_vulnerabilities.json"
-
-# Load JSON data
-with open(path, "r", encoding="utf-8") as f:
-    data = json.load(f)
-
-vuln_types = []
-
-for entry in data:
-    title = entry.get("title")
-    if not isinstance(title, str):
-        continue
-
-    # Split on the first dash (" - ") and take the right-hand side
-    if " - " in title:
-        vuln = title.split(" - ", 1)[1].strip()
-        vuln_types.append(vuln)
-
-# Count occurrences
-counts = Counter(vuln_types)
-top25 = counts.most_common(25)
-total = sum(counts.values())
-
-if not vuln_types:
-    print("error")
-    exit()
-
-# Print table of top 25
-print("\nTop 25 Vulnerability Types:\n")
-for name, count in top25:
-    rel = (count / total) * 100
-    print(f"{name:60} {count:5d} ({rel:5.2f}%)")
-
-# Visualization
-labels, values = zip(*top25)
-plt.figure(figsize=(10, 6))
-plt.barh(labels[::-1], values[::-1], color="cornflowerblue")
-plt.title("Top 25 Most Common Vulnerability Types")
-plt.xlabel("Count")
-plt.ylabel("Vulnerability Type")
-plt.tight_layout()
-plt.show()
-

requirements.txt

@@ -1 +0,0 @@
-requests

scripts/CVE_scanner_new.py

@@ -1,89 +0,0 @@
-#!/usr/bin/env python3
-import json
-import os
-
-# vars
-JSON_FILE = "wordfence_vulnerabilities.json"
-BAD_LIST = []
-
-JSON_OUT = "matches.json"
-
-
-def load_local(path):
-    with open(path, "r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def get_entries(data):
-    # try to find the main list of vulnerabilities
-    if isinstance(data, dict):
-        if "vulnerabilities" in data:
-            return data["vulnerabilities"]
-        for value in data.values():
-            if isinstance(value, list):
-                return value
-        return []
-    elif isinstance(data, list):
-        return data
-    else:
-        return []
-
-
-def main():
-    if os.path.exists(JSON_FILE):
-        print("Loading ...", JSON_FILE)
-        data = load_local(JSON_FILE)
-    else:
-        print("Error: JSON file not found:", JSON_FILE)
-        return
-
-    entries = get_entries(data)
-
-    bad_set = set(BAD_LIST)
-    hits = []
-
-    for item in entries:
-        if not isinstance(item, dict):
-            continue
-
-        if item.get("type") != "plugin":
-            continue
-
-        slug = (item.get("slug") or "").strip()
-        if not slug:
-            continue
-
-        if slug in bad_set:
-            hit = {
-                "slug": slug,
-                "title": item.get("title"),
-                "cve": item.get("cve"),
-                "severity": item.get("severity"),
-                "cvss_score": item.get("cvss_score"),
-                "cve_link": item.get("cve_link"),
-                "raw": item
-            }
-            hits.append(hit)
-
-    if not hits:
-        print("No matches at all")
-    else:
-        print("Found", len(hits), "matches:")
-        for h in hits:
-            print(
-                "-",
-                h["slug"],
-                "| CVE:", h.get("cve"),
-                "| sev:", h.get("severity"),
-                "| CVSS:", h.get("cvss_score")
-            )
-            print("  link:", h.get("cve_link"))
-
-    # save
-    with open(JSON_OUT, "w", encoding="utf-8") as f:
-        json.dump(hits, f, ensure_ascii=False, indent=2)
-    print("Saved to", JSON_OUT)
-
-
-if __name__ == "__main__":
-    main()

scripts/IMPORTANT.txt

@@ -1,2 +0,0 @@
-get_CVE.py **MUST** be ran before cve_scanner_new1 can be ran since it runs using a list that is fetched daily.
-CVE_scanner.py runs on a static list called wordfence_vulnerabilities.json that is stored in the same directory, which is an old list of major vulnerabilities.

scripts/cve_scanner_new1.py

@@ -1,77 +0,0 @@
-#!/usr/bin/env python3
-import json
-import os
-
-JSON_FILE = "wordfence_CVE.json"
-JSON_OUT = "malware.json"
-BAD_PLUGINS = []
-
-
-def load_file(path):
-    with open(path, "r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def main():
-    if not os.path.exists(JSON_FILE):
-        print("json file not found:", JSON_FILE)
-        return
-
-    data = load_file(JSON_FILE)
-
-    bad_set = set()
-    for n in BAD_PLUGINS:
-        if isinstance(n, str):
-            n = n.strip()
-            if n:
-                bad_set.add(n)
-
-    hits = []
-
-    for key, item in data.items():
-        if not isinstance(item, dict):
-            continue
-
-        software_list = item.get("software") or []
-        for s in software_list:
-            if not isinstance(s, dict):
-                continue
-            if s.get("type") != "plugin":
-                continue
-
-            name = (s.get("name") or "").strip()
-            if name not in bad_set:
-                continue
-
-            affected = s.get("affected_versions") or {}
-            if not affected:
-                continue
-
-            refs = item.get("references") or []
-            ref = refs[0] if refs else None
-
-            hit = {
-                "id": item.get("id"),
-                "title": item.get("title"),
-                "name": name,
-                "slug": s.get("slug"),
-                "affected_versions": list(affected.keys()),
-                "link": ref,
-                "raw": item
-            }
-            hits.append(hit)
-
-    if not hits:
-        print("no matches found")
-    else:
-        print("found", len(hits), "matches")
-        for h in hits:
-            print("-", h["name"], "|", ", ".join(h["affected_versions"]))
-
-    with open(JSON_OUT, "w", encoding="utf-8") as f:
-        json.dump(hits, f, ensure_ascii=False, indent=2)
-    print("saved json to", JSON_OUT)
-
-
-if __name__ == "__main__":
-    main()

scripts/get_CVE.py

@@ -1,22 +0,0 @@
-#!/usr/bin/env python3
-import requests
-import os
-
-URL = "https://www.wordfence.com/api/intelligence/v2/vulnerabilities/scanner"
-OUT_FILE = "wordfence_CVE.json"
-
-def main():
-    print("API request.....")
-    r = requests.get(URL, timeout=60)
-    r.raise_for_status()
-
-    script_dir = os.path.dirname(os.path.abspath(__file__))
-    out_path = os.path.join(script_dir, OUT_FILE)
-
-    with open(out_path, "w", encoding="utf-8") as f:
-        f.write(r.text)
-
-    print("File Saved!")
-
-if __name__ == "__main__":
-    main()