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)