#!/usr/bin/env python3
import argparse
import json
import smtplib
from collections import defaultdict
from datetime import datetime, timezone
from email.message import EmailMessage
from pathlib import Path
def parse_env(env_path: Path) -> dict:
values = {}
if not env_path.exists():
return values
for raw_line in env_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip().strip('"').strip("'")
return values
def choose_date(arg_value: str | None) -> str:
if arg_value:
return arg_value
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
def load_events(log_path: Path) -> list[dict]:
events = []
if not log_path.exists():
return events
for line in log_path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
events.append(json.loads(line))
except json.JSONDecodeError:
continue
return events
def score_events(events: list[dict], env: dict) -> tuple[dict, list[dict]]:
suspicious_weights = {
"web_login_failed": 5,
"api_login_failed": 5,
"password_change_failed": 2,
"path_traversal": 2,
"env_probe": 40,
"shell_probe": 5,
"proxy_vpn_detected": 8,
"missing_s_token": 8,
"invalid_s_token": 8,
"stale_s_token": 6,
"ip_mismatch_s_token": 8,
"manual_block": 8,
"input_blocked": 30,
}
request_error_weight = 1
threshold = int(env.get("SECURITY_PERSISTENT_BLOCK_THRESHOLD", "12"))
scores = defaultdict(lambda: {"score": 0, "events": [], "request_errors": 0, "event_counts": defaultdict(int)})
blocked_entries = []
for event in events:
ip = event.get("ip")
if not ip:
continue
if event.get("user_id") not in (None, "", 0):
continue
event_name = event.get("event", "")
status = int(event.get("status", 200) or 200)
if event_name == "request" and status >= 400:
scores[ip]["score"] += request_error_weight
scores[ip]["request_errors"] += 1
if event_name in suspicious_weights:
scores[ip]["score"] += suspicious_weights[event_name]
scores[ip]["events"].append(event_name)
scores[ip]["event_counts"][event_name] += 1
now = datetime.now(timezone.utc)
for ip, summary in scores.items():
if summary["score"] < threshold:
continue
dominant = sorted(
summary["event_counts"].items(),
key=lambda item: item[1],
reverse=True,
)[:3]
dominant_reason = ", ".join(f"{name}x{count}" for name, count in dominant) or "request_errors"
blocked_entries.append({
"ip": ip,
"reason": f"python-analyzer score={summary['score']} dominant={dominant_reason}",
"blocked_at": now.isoformat(),
"score": summary["score"],
"request_errors": summary["request_errors"],
})
return scores, blocked_entries
def write_blocklist(blocklist_path: Path, blocked_entries: list[dict]) -> None:
blocklist_path.parent.mkdir(parents=True, exist_ok=True)
existing_entries = []
if blocklist_path.exists():
try:
decoded = json.loads(blocklist_path.read_text(encoding="utf-8"))
existing_entries = decoded.get("blocked_ips", []) if isinstance(decoded, dict) else []
except json.JSONDecodeError:
existing_entries = []
merged = {entry.get("ip"): entry for entry in existing_entries if isinstance(entry, dict) and entry.get("ip")}
for entry in blocked_entries:
merged[entry["ip"]] = entry
payload = {"blocked_ips": list(merged.values())}
blocklist_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
def html_escape(value) -> str:
return (
str(value)
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
def build_summary_html(date_value: str, body: dict) -> str:
blocked_count = int(body.get("blocked_count", 0) or 0)
top_scores = body.get("top_scores", [])
blocked_ips = body.get("blocked_ips", [])
header_color = "#dc3545" if blocked_count > 0 else "#007bff"
status_label = "Blocked IPs detected" if blocked_count > 0 else "No IPs blocked"
top_score_rows = ""
for item in top_scores:
top_score_rows += f"""
| {html_escape(item.get("ip", "Unknown"))} |
{html_escape(item.get("score", 0))} |
{html_escape(item.get("request_errors", 0))} |
"""
if not top_score_rows:
top_score_rows = """
| No suspicious scores found. |
"""
blocked_rows = ""
for item in blocked_ips:
blocked_rows += f"""
| {html_escape(item.get("ip", "Unknown"))} |
{html_escape(item.get("reason", "Blocked by analyzer"))} |
"""
if not blocked_rows:
blocked_rows = """
| No IPs were added to the blocklist. |
"""
return f"""
RSC Security Analyzer Summary
Security Analyzer Summary
{html_escape(date_value)} - {html_escape(status_label)}
|
|
Hello Harry,
RSC Smart WAF analyzer completed the daily security scan. Review the summary below for blocked IPs and suspicious request scores.
|
Date: {html_escape(date_value)}
Blocked Count: {blocked_count}
Status: {html_escape(status_label)}
|
Top Suspicious Scores
| IP Address |
Score |
Errors |
{top_score_rows}
Blocked IPs
| IP Address |
Reason |
{blocked_rows}
|
|
© 2026 RSC Stock Audit
|
|
"""
def send_email(env: dict, date_value: str, blocked_entries: list[dict], scores: dict) -> None:
recipient = env.get("SECURITY_ALERT_EMAIL") or env.get("MAIL_FROM_ADDRESS")
host = env.get("MAIL_HOST")
port = int(env.get("MAIL_PORT", "25"))
username = env.get("MAIL_USERNAME")
password = env.get("MAIL_PASSWORD")
sender = env.get("MAIL_FROM_ADDRESS", "security@example.com")
body = {
"date": date_value,
"blocked_count": len(blocked_entries),
"blocked_ips": blocked_entries,
"top_scores": sorted(
(
{"ip": ip, "score": detail["score"], "request_errors": detail["request_errors"]}
for ip, detail in scores.items()
if detail["score"] > 0
),
key=lambda item: item["score"],
reverse=True,
)[:10],
}
if not recipient:
return
if not host:
print(json.dumps({"mail": "skipped", "reason": "MAIL_HOST not configured", "summary": body}, ensure_ascii=False))
return
message = EmailMessage()
message["Subject"] = f"RSC security analyzer summary {date_value}"
message["From"] = sender
message["To"] = recipient
message.set_content(json.dumps(body, indent=2))
message.add_alternative(build_summary_html(date_value, body), subtype="html")
try:
with smtplib.SMTP(host, port, timeout=20) as smtp:
if env.get("MAIL_ENCRYPTION") == "tls":
smtp.starttls()
if username:
smtp.login(username, password or "")
smtp.send_message(message)
except Exception as exc:
print(json.dumps({
"mail": "failed",
"reason": str(exc),
"recipient": recipient,
}, ensure_ascii=False))
def main() -> int:
parser = argparse.ArgumentParser(description="Analyze Laravel security JSONL logs and refresh blocked IPs.")
parser.add_argument("--project-root", required=True)
parser.add_argument("--date")
args = parser.parse_args()
project_root = Path(args.project_root).resolve()
env = parse_env(project_root / ".env")
date_value = choose_date(args.date)
log_path = project_root / "storage" / "logs" / "security" / f"security-{date_value}.jsonl"
blocklist_path = project_root / "storage" / "app" / "security" / "blocked_ips.json"
events = load_events(log_path)
scores, blocked_entries = score_events(events, env)
write_blocklist(blocklist_path, blocked_entries)
send_email(env, date_value, blocked_entries, scores)
print(json.dumps({
"date": date_value,
"log_file": str(log_path),
"events_read": len(events),
"blocked_count": len(blocked_entries),
"blocklist_file": str(blocklist_path),
}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())