diff --git a/WordPress-Metrics/README.md b/WordPress-Metrics/README.md new file mode 100644 index 0000000..6e2e1b1 --- /dev/null +++ b/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} \ No newline at end of file diff --git a/WordPress-Metrics/analyze_data.py b/WordPress-Metrics/analyze_data.py new file mode 100644 index 0000000..e2f2e1d --- /dev/null +++ b/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() \ No newline at end of file diff --git a/WordPress-Metrics/analyze_plugins.py b/WordPress-Metrics/analyze_plugins.py new file mode 100644 index 0000000..1651802 --- /dev/null +++ b/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() diff --git a/WordPress-Metrics/query_plugin.py b/WordPress-Metrics/query_plugin.py new file mode 100644 index 0000000..79c53d2 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/WordPress-Metrics/scrape_wordpress_plugins.py b/WordPress-Metrics/scrape_wordpress_plugins.py new file mode 100644 index 0000000..a029c7f --- /dev/null +++ b/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) \ No newline at end of file diff --git a/WordPress-Metrics/security_rating.py b/WordPress-Metrics/security_rating.py new file mode 100644 index 0000000..3c1c6ea --- /dev/null +++ b/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 ") + sys.exit(1) + slug = sys.argv[1] + rate_plugin(slug)