Archived
Compare commits
8
Commits
plugin-list
...
metrics
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c3d861cb51
|
||
|
|
205bafb94b | ||
|
|
6b3c1e2bd6 | ||
|
|
d18b433d01 | ||
|
|
01e86c78d0 | ||
|
|
0b40e4e799 | ||
|
|
7b1217a2ef | ||
|
|
30395938ed |
@@ -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}
|
||||||
@@ -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()
|
||||||
@@ -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()
|
||||||
@@ -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
|
||||||
@@ -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)
|
||||||
@@ -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)
|
||||||
Reference in New Issue
Block a user