diff --git "a/backend/scanners_core.py" "b/backend/scanners_core.py" new file mode 100644--- /dev/null +++ "b/backend/scanners_core.py" @@ -0,0 +1,3755 @@ +import sys +import os +import re +import time +import json +import math +import uuid +import html +import io +import hashlib +import sqlite3 +import socket +import ssl +import base64 +import bcrypt +import jwt +import requests +import urllib3 +import ipaddress +import queue +import threading +import statistics +import itertools +import traceback +import concurrent.futures +from datetime import datetime, timezone, timedelta +from typing import Any, Callable, Literal +from collections import defaultdict +from functools import wraps +from urllib.parse import urljoin, urlparse +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +BACKEND_DIR = os.path.join(BASE_DIR, 'backend') +if BASE_DIR not in sys.path: + sys.path.insert(0, BASE_DIR) +if BACKEND_DIR not in sys.path: + sys.path.insert(0, BACKEND_DIR) + +from bs4 import BeautifulSoup +from celery import Celery +from celery.schedules import crontab +from scanners.base_scanner import ( + active_scan_logs, add_log, get_scan_logs, parse_domain, + cleanup_scan_logs, schedule_log_cleanup, emit_scan_progress +) +from scanners import get_pipeline, get_phases, build_scanner, apply_scan_options +try: + from backend.utils.fuzzer_engine import ContextAwareFuzzer +except ImportError: + from utils.fuzzer_engine import ContextAwareFuzzer +from cryptography import x509 +from cryptography.hazmat.backends import default_backend + +import stripe +from flask import ( + Flask, Blueprint, request, jsonify, current_app, send_from_directory, + send_file, render_template, abort, g, Response, make_response +) +from werkzeug.utils import secure_filename +from flask_cors import CORS +from flask_limiter import Limiter +from flask_limiter.util import get_remote_address +from flask_socketio import SocketIO, emit, join_room, leave_room +from flask_sqlalchemy import SQLAlchemy +from markupsafe import escape + +from reportlab.lib import colors +from reportlab.lib.pagesizes import letter +from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle +from reportlab.pdfgen import canvas +from reportlab.platypus import ( + SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak, Image, Flowable, KeepTogether +) +from reportlab.graphics.shapes import Drawing +from reportlab.graphics.charts.barcharts import VerticalBarChart + +from sqlalchemy import event, func, inspect, text +from sqlalchemy.engine import Engine + +try: + from backend.utils.email_service import ( + send_welcome_email, + send_scan_started, + send_scan_completed, + send_scan_failed, + send_critical_alert + ) +except ImportError: + from utils.email_service import ( + send_welcome_email, + send_scan_started, + send_scan_completed, + send_scan_failed, + send_critical_alert + ) + +from .extensions import db, celery, socketio, limiter +from .models import * + + +# --- From scanner.py --- +""" +scanner.py - Scan orchestration engine +======================================= +Fixes applied (June 2026): + FIX-1: Celery import wrapped in try/except - backend works without Redis + FIX-2: _run_scan_job() is a plain function called directly from threads + FIX-3: Each DB write uses a fresh session, properly removed after use + FIX-4: SQLAlchemy scoped_session used for thread-safe DB access + FIX-5: Proper error handling ensures scan always marks as failed/completed + BUG-6 FIX: cleanup_scan_logs() deferred 5 min post-completion via + schedule_log_cleanup() - prevents race with frontend /logs polling + ENH: Deduplication of vulnerabilities before DB write + ENH: Scan timeout enforcement (SCANNER_TIMEOUT_SECONDS) +""" + + + +# ── Celery is optional - works without Redis/Celery installed ──────────────── +try: + from .config import _is_redis_running, Config + CELERY_AVAILABLE = _is_redis_running(Config.CELERY_BROKER_URL) +except Exception: + CELERY_AVAILABLE = False + +# Global timeout: 2 hours for Deep scan (was 600s = too short for 80+ modules) +SCANNER_TIMEOUT_SECONDS = 7200 + + +def _clean_nul(val) -> str: + if val is None: + return "" + if not isinstance(val, str): + val = str(val) + return val.replace("\x00", "").replace("\u0000", "") + + +def calculate_security_score_from_counts(counts: dict) -> int: + crit = counts.get("critical", 0) or counts.get("Critical", 0) + high = counts.get("high", 0) or counts.get("High", 0) + med = counts.get("medium", 0) or counts.get("Medium", 0) + low = counts.get("low", 0) or counts.get("Low", 0) + + if crit == 0 and high == 0 and med == 0 and low == 0: + return 100 + + # Critical penalty: 1st=15, 2nd=10, 3rd-5th=6, 6th-10th=3, 11th+=1 + crit_deduction = 0 + if crit > 0: crit_deduction += 15 + if crit > 1: crit_deduction += 10 + if crit > 2: crit_deduction += min(crit - 2, 3) * 6 + if crit > 5: crit_deduction += min(crit - 5, 5) * 3 + if crit > 10: crit_deduction += (crit - 10) * 1 + + # High penalty: 1st=7, 2nd-5th=4, 6th-10th=2, 11th+=0.5 + high_deduction = 0 + if high > 0: high_deduction += 7 + if high > 1: high_deduction += min(high - 1, 4) * 4 + if high > 5: high_deduction += min(high - 5, 5) * 2 + if high > 10: high_deduction += (high - 10) * 0.5 + + # Medium penalty: 1st-3rd=3, 4th-8th=1.5, 9th+=0.5 + med_deduction = 0 + if med > 0: med_deduction += min(med, 3) * 3 + if med > 3: med_deduction += min(med - 3, 5) * 1.5 + if med > 8: med_deduction += (med - 8) * 0.5 + + # Low penalty: 1st-5th=1, 6th+=0.25 + low_deduction = 0 + if low > 0: low_deduction += min(low, 5) * 1 + if low > 5: low_deduction += (low - 5) * 0.25 + + total_deduction = crit_deduction + high_deduction + med_deduction + low_deduction + return max(0, min(100, int(round(100 - total_deduction)))) + + +def calculate_security_score(vulns: list[dict]) -> int: + if not vulns: + return 100 + + counts = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0} + for v in vulns: + sev = v.get("severity", "Low") + if sev in counts: + counts[sev] += 1 + + return calculate_security_score_from_counts(counts) + + +def _deduplicate_scan_vulns(vulns: list[dict]) -> list[dict]: + """ + Cross-scanner deduplication of vulnerability findings. + Dedup key: (title, category). Keeps the highest-confidence entry. + ENH: Prevents DB flooding with identical findings from multiple scanners. + """ + conf_rank = {"Low": 0, "Medium": 1, "High": 2, "Confirmed": 3} + seen: dict[tuple, dict] = {} + for v in vulns: + key = (v.get("title", ""), v.get("category", "")) + if key not in seen: + seen[key] = v + else: + existing_rank = conf_rank.get(seen[key].get("confidence", "Low"), 0) + new_rank = conf_rank.get(v.get("confidence", "Low"), 0) + if new_rank > existing_rank: + seen[key] = v + return list(seen.values()) + + +# ── Core scan job - plain Python function, no Celery dependency ────────────── + +def _run_scan_job(scan_id: str) -> None: + """ + Main scan pipeline executor. + MUST be called inside an active Flask app context. + Uses db.session with proper cleanup between writes. + """ + try: + # Set socketio instance for real-time progress updates + from scanners.base_scanner import set_socketio_instance + if hasattr(current_app, 'socketio'): + set_socketio_instance(current_app.socketio) + + # Refresh the session to get a clean state for this thread + db.session.remove() + + scan = db.session.get(Scan, scan_id) + if not scan: + print(f"[Scanner] Scan {scan_id} not found in database.", flush=True) + return + + target = scan.target_url + scan_type = scan.scan_type + domain = parse_domain(target) + + # Per-module timeout per scan intensity (Deep gets 600s = 10 min per module) + MODULE_TIMEOUTS = { + "quick": 60, + "standard": 120, + "advanced": 180, + "deep": 600, + } + _module_timeout = MODULE_TIMEOUTS.get((scan_type or "standard").lower(), 120) + + add_log(scan_id, "INFO", f"LarShield v2.0 - {scan_type.upper()} SCAN INITIATED") + add_log(scan_id, "INFO", f"Target: {target}") + add_log(scan_id, "INFO", f"Domain: {domain}") + add_log(scan_id, "INFO", f"Scan ID: {scan_id}") + + # Mark as scanning + try: + db.session.remove() + scan = db.session.get(Scan, scan_id) + if scan: + scan.status = "scanning" + try: + scan.ssl_info = get_ssl_info(target) + except Exception as ssl_e: + print(f"[Scanner] Failed to cache SSL info: {ssl_e}") + db.session.commit() + except Exception as e: + db.session.rollback() + add_log(scan_id, "WARNING", f"Could not update scan status: {e}") + finally: + db.session.remove() + + # Build the scanner pipeline + db.session.remove() + scan = db.session.get(Scan, scan_id) + scan_options = getattr(scan, 'scan_options', None) + auth_headers = getattr(scan, 'auth_headers', None) + db.session.remove() + + pipeline = apply_scan_options( + get_pipeline(scan_type), scan_type, scan_options + ) + + if scan_options: + add_log(scan_id, "INFO", + f"Advanced options - crawl depth: {scan_options.get('crawl_depth', 'default')}, " + f"exclusions: {len(scan_options.get('exclude_paths', []))}, " + f"red-team: {scan_options.get('enable_red_team', False)}") + + all_vulns: list[dict] = [] + + def run_scanner_step(step_num, step_name, scanner_cls, kwargs, total_steps): + add_log(scan_id, "INFO", + f"Step {step_num}/{total_steps}: Running {scanner_cls.SCANNER_NAME}...") + try: + scanner = build_scanner( + step_name, scanner_cls, kwargs, + scan_id=scan_id, target=target, + domain=domain, auth_headers=auth_headers, + ) + import concurrent.futures as _cf_inner + with _cf_inner.ThreadPoolExecutor(max_workers=1) as _inner_exec: + _fut = _inner_exec.submit(scanner.run) + try: + step_vulns = _fut.result(timeout=_module_timeout) or [] + except _cf_inner.TimeoutError: + add_log(scan_id, "WARNING", + f"[{step_name}] MODULE TIMEOUT after {_module_timeout}s - skipped.") + try: + _fut.cancel() + except Exception: + pass + return [] + n = len(step_vulns) if step_vulns else 0 + add_log(scan_id, + "SUCCESS" if not step_vulns else "WARNING", + f"[{step_name}] Completed - {n} finding(s).") + return step_vulns or [] + except Exception as e: + add_log(scan_id, "WARNING", + f"[{step_name}] Scanner raised an unexpected exception: {e}") + return [] + + _scan_start_time = time.time() + + # ── Universal Phase-Based Execution Engine ───────────────────────────── + # Runs for ALL scan types: Quick (2 phases), Advanced (4 phases), Deep (8 phases) + # Each phase: modules run CONCURRENTLY (up to max_workers_per_phase) + # Phases run SEQUENTIALLY — ensures recon finishes before injection probing, etc. + # ────────────────────────────────────────────────────────────────────── + + # Per-scan-type concurrency caps per phase + MAX_WORKERS_PER_PHASE = { + "quick": 6, + "standard": 8, + "advanced": 8, + "deep": 8, + "ssl": 4, + "port": 2, + } + _max_workers = MAX_WORKERS_PER_PHASE.get((scan_type or "advanced").lower(), 8) + + # Wall-clock hard limits per scan type (seconds) + HARD_LIMITS = { + "quick": 300, # 5 min + "advanced": 3600, # 1 hour + "standard": 3600, + "deep": 21600, # 6 hours + "ssl": 300, + "port": 600, + } + _hard_limit = HARD_LIMITS.get((scan_type or "advanced").lower(), 3600) + + phases = get_phases(scan_type) + total_steps = len(pipeline) + n_phases = len(phases) + + add_log(scan_id, "INFO", + f"[{scan_type} Scan] Starting phase-based execution: " + f"{total_steps} modules across {n_phases} phase(s), " + f"max {_max_workers} concurrent per phase, " + f"{_module_timeout}s per module timeout.") + + # Build name → (i, name, cls, kwargs) lookup from the pipeline + pipeline_lookup: dict = {} + for i, (name, cls, kwargs) in enumerate(pipeline): + pipeline_lookup[name] = (i, name, cls, kwargs) + + assigned_names: set = set() + + for phase_idx, phase in enumerate(phases, 1): + # Collect steps for this phase that exist in the pipeline and aren't already run + phase_steps = [ + pipeline_lookup[n] + for n in phase["keys"] + if n in pipeline_lookup and n not in assigned_names + ] + for step in phase_steps: + assigned_names.add(step[1]) # mark as assigned + + if not phase_steps: + add_log(scan_id, "INFO", + f"[{scan_type}] {phase['name']} — no matching modules, skipping.") + continue + + # Hard-limit wall-clock check + elapsed_total = time.time() - _scan_start_time + if elapsed_total > _hard_limit: + add_log(scan_id, "WARNING", + f"[{scan_type}] Hard time limit ({_hard_limit}s) reached " + f"before {phase['name']}. Stopping early.") + break + + add_log(scan_id, "INFO", + f"[{scan_type}] ▶ {phase['name']} " + f"({len(phase_steps)} module(s), phase {phase_idx}/{n_phases})...") + + phase_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(phase_steps), _max_workers) + ) + phase_futures = [ + phase_executor.submit( + run_scanner_step, i + 1, name, cls, kwargs, total_steps + ) + for i, name, cls, kwargs in phase_steps + ] + # Phase timeout = modules × per-module timeout, capped at 30 min + phase_timeout = min(len(phase_steps) * _module_timeout, 1800) + try: + for future in concurrent.futures.as_completed(phase_futures, + timeout=phase_timeout): + try: + result = future.result() + if result: + all_vulns.extend(result) + except Exception: + pass + except concurrent.futures.TimeoutError: + add_log(scan_id, "WARNING", + f"[{scan_type}] {phase['name']} timed out after " + f"{phase_timeout}s — continuing to next phase.") + finally: + try: + phase_executor.shutdown(wait=False, cancel_futures=True) + except TypeError: + phase_executor.shutdown(wait=False) + + # Run any pipeline modules that weren't assigned to any phase + remaining_steps = [ + (i, name, cls, kwargs) + for i, (name, cls, kwargs) in enumerate(pipeline) + if name not in assigned_names + ] + if remaining_steps: + add_log(scan_id, "INFO", + f"[{scan_type}] Running {len(remaining_steps)} unassigned module(s)...") + rem_executor = concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(remaining_steps), _max_workers) + ) + rem_futures = [ + rem_executor.submit( + run_scanner_step, i + 1, name, cls, kwargs, total_steps + ) + for i, name, cls, kwargs in remaining_steps + ] + rem_timeout = min(len(remaining_steps) * _module_timeout, 1800) + try: + for future in concurrent.futures.as_completed(rem_futures, + timeout=rem_timeout): + try: + result = future.result() + if result: + all_vulns.extend(result) + except Exception: + pass + except concurrent.futures.TimeoutError: + add_log(scan_id, "WARNING", + f"[{scan_type}] Unassigned modules timed out after {rem_timeout}s.") + finally: + try: + rem_executor.shutdown(wait=False, cancel_futures=True) + except TypeError: + rem_executor.shutdown(wait=False) + + # ENH: Cross-scanner deduplication before scoring and DB write + original_count = len(all_vulns) + all_vulns = _deduplicate_scan_vulns(all_vulns) + if original_count != len(all_vulns): + add_log(scan_id, "INFO", + f"Deduplication: {original_count} → {len(all_vulns)} unique findings.") + + score = calculate_security_score(all_vulns) + + add_log(scan_id, "INFO", f"Running AI post-processing on {len(all_vulns)} finding(s)...") + self_metadata: list[dict] = [] + try: + + tech_fingerprints: list[dict] = [] + for v in all_vulns: + resp_det = v.get("response_details", "") + headers = {"server": v.get("server_header", ""), "x-powered-by": v.get("powered_by", "")} + if resp_det: + tech_fingerprints.extend(match_tech(resp_det, headers)) + + if tech_fingerprints: + unique_tech = {} + for t in tech_fingerprints: + unique_tech[t["name"]] = t + for t in unique_tech.values(): + cves = find_cves(t["name"], t.get("version")) + t["matched_cves"] = cves + self_metadata.append({"type": "tech", "data": t}) + + chains = detect_chains(all_vulns) + for chain in chains: + add_log(scan_id, "CRITICAL", + f"[CHAIN] {chain['chain_name']} (CVSS {chain['cvss_score']})") + self_metadata.append({"type": "chain", "data": chain}) + + high_confidence = [v for v in all_vulns + if v.get("confidence") in ("Confirmed", "High")] + for v in high_confidence[:5]: + try: + exploit = generate_exploit(v) + v["exploit_poc"] = exploit + v["remediation_code"] = generate_remediation(v) + except Exception: + pass + + add_log(scan_id, "INFO", + f"AI engine: {len(self_metadata)} metadata items, " + f"{len(high_confidence)} high-conf findings enriched.") + except Exception as ai_err: + add_log(scan_id, "INFO", f"AI enrichment (non-fatal): {ai_err}") + + add_log(scan_id, "INFO", f"Syncing {len(all_vulns)} finding(s) to database...") + + # ── Write results to DB - fresh session per write ───────────────────── + try: + db.session.remove() + + # Persist each vulnerability + for v_data in all_vulns: + try: + vuln = Vulnerability( + scan_id=scan_id, + title=_clean_nul(v_data.get("title", "")), + severity=_clean_nul(v_data.get("severity", "Low")), + category=_clean_nul(v_data.get("category", "")), + description=_clean_nul(v_data.get("description", "")), + remediation=_clean_nul(v_data.get("remediation", "")), + cvss_score=float(v_data.get("cvss_score", 0)), + evidence=_clean_nul(v_data.get("evidence", "")), + payload=_clean_nul(v_data.get("payload", "")), + request_details=_clean_nul(v_data.get("request_details", "")), + response_details=_clean_nul(v_data.get("response_details", "")), + cwe_ids=v_data.get("cwe_ids"), + owasp_category=_clean_nul(v_data.get("owasp_category")), + exploit_poc=_clean_nul(v_data.get("exploit_poc")), + remediation_code=_clean_nul(v_data.get("remediation_code")), + ) + db.session.add(vuln) + except Exception as ve: + add_log(scan_id, "WARNING", f"Could not create vuln record: {ve}") + + # Update scan status + scan = db.session.get(Scan, scan_id) + if scan: + scan.status = "completed" + scan.security_score = score + scan.completed_at = datetime.now(timezone.utc) + + db.session.commit() + + try: + emit_scan_progress(scan_id, 'scan_progress', {'status': 'completed'}) + except Exception: + pass + + crit = sum(1 for v in all_vulns if v.get("severity") == "Critical") + high = sum(1 for v in all_vulns if v.get("severity") == "High") + med = sum(1 for v in all_vulns if v.get("severity") == "Medium") + low = sum(1 for v in all_vulns if v.get("severity") == "Low") + + add_log(scan_id, "SUCCESS", + f"SCAN COMPLETE - Security Score: {score}/100") + add_log(scan_id, "SUCCESS", + f"Findings: {crit} Critical | {high} High | {med} Medium | {low} Low") + add_log(scan_id, "SUCCESS", + f"Total unique vulnerabilities: {len(all_vulns)}") + + try: + if scan: + scan_user = db.session.get(User, scan.user_id) + if scan_user: + duration_secs = (datetime.utcnow() - scan.started_at).total_seconds() if scan.started_at else 0 + duration_str = f"{int(duration_secs // 60)}m {int(duration_secs % 60)}s" if duration_secs > 0 else "< 1m" + send_scan_completed( + scan_user.email, + scan_user.email.split('@')[0].capitalize(), + scan.target_url, + duration_str, + str(len(all_vulns)), + f"https://wss.larshield.com/dashboard/scans/{scan.id}", + str(crit), + str(high), + str(med), + str(low) + ) + + # Suggestion 4: Send critical alert to Org Admin if high/critical vulns found + if crit > 0 or high > 0: + org_admin = User.query.filter_by(org_id=scan.org_id, role='org_admin').first() + if org_admin and org_admin.id != scan.user_id: # Only if they aren't the one who just got the completed email + send_critical_alert( + org_admin.email, + org_admin.first_name or org_admin.email.split('@')[0].capitalize(), + scan.target_url, + duration_str, + str(len(all_vulns)), + f"https://wss.larshield.com/dashboard/scans/{scan.id}", + str(crit), + str(high), + str(med), + str(low) + ) + print(f"[Email] Critical Alert sent to Org Admin: {org_admin.email}") + + except Exception as e: + print(f"[Email] Failed to send scan completed/alert email: {e}") + + # ── Webhook alert ───────────────────────────────────────────────── + try: + alert_settings = AlertSettings.query.filter_by( + user_id=scan.user_id + ).first() if scan else None + + # Collect all webhook URLs to notify + urls_to_notify = [] + + if alert_settings and alert_settings.webhook_url: + send = False + thresh = alert_settings.severity_threshold + if thresh == "All": send = True + elif thresh == "Critical" and crit > 0: send = True + elif thresh == "High" and (crit > 0 or high > 0): send = True + elif thresh == "Medium" and (crit > 0 or high > 0 or med > 0): send = True + else: send = True + if send: + urls_to_notify.append(alert_settings.webhook_url) + + # Organization-level webhook (if high/crit found) + if scan and scan.org_id: + org = db.session.get(Organization, scan.org_id) + if org and org.webhook_url and (crit > 0 or high > 0): + if org.webhook_url not in urls_to_notify: + urls_to_notify.append(org.webhook_url) + + if urls_to_notify: + db_vulns = Vulnerability.query.filter_by(scan_id=scan_id).all() + for url in urls_to_notify: + try: + send_webhook_alert(url, scan, db_vulns, crit, high) + except Exception as inner_e: + add_log(scan_id, "WARNING", f"Failed to send webhook to {url}: {inner_e}") + + add_log(scan_id, "INFO", "[System] Webhook alerts dispatched successfully.") + except Exception as we: + add_log(scan_id, "WARNING", f"Webhook error (non-fatal): {we}") + + except Exception as db_err: + db.session.rollback() + add_log(scan_id, "CRITICAL", f"Database write failure: {str(db_err)}") + print(f"[Scanner] DB write error for scan {scan_id}: {db_err}", flush=True) + + # Mark scan as failed + try: + db.session.remove() + scan = db.session.get(Scan, scan_id) + if scan: + scan.status = "failed" + db.session.commit() + try: + scan_user = db.session.get(User, scan.user_id) + if scan_user: + send_scan_failed( + scan_user.email, + scan_user.email.split('@')[0].capitalize(), + scan.target_url, + scan.scan_type, + str(db_err) + ) + except Exception as e: + print(f"[Email] Failed to send scan failed email: {e}") + try: + emit_scan_progress(scan_id, 'scan_progress', {'status': 'failed'}) + except Exception: + pass + except Exception: + db.session.rollback() + + except Exception as fatal_err: + print(f"[Scanner] Fatal error in scan {scan_id}: {fatal_err}", flush=True) + traceback.print_exc() + # Mark as failed + try: + db.session.remove() + scan = db.session.get(Scan, scan_id) + if scan: + scan.status = "failed" + db.session.commit() + try: + scan_user = db.session.get(User, scan.user_id) + if scan_user: + send_scan_failed( + scan_user.email, + scan_user.email.split('@')[0].capitalize(), + scan.target_url, + scan.scan_type, + "Fatal system error" + ) + except Exception as e: + print(f"[Email] Failed to send scan failed email: {e}") + try: + emit_scan_progress(scan_id, 'scan_progress', {'status': 'failed'}) + except Exception: + pass + except Exception: + db.session.rollback() + finally: + # Always clean up the session after completion + try: + db.session.remove() + except Exception: + pass + # BUG-6 FIX: Defer log cleanup by 5 minutes so frontend /logs polling works. + # Old code: cleanup_scan_logs(scan_id) - deleted logs while scan appeared "scanning" + schedule_log_cleanup(scan_id, delay_seconds=300) + + +# ── Celery tasks (optional - only registered if Celery is available) ───────── + +if CELERY_AVAILABLE and celery: + @celery.task(bind=True, name="run_background_scan") + def run_background_scan_task(self, scan_id: str) -> None: + """Celery task wrapper - used when Redis is available.""" + app = create_app() + with app.app_context(): + _run_scan_job(scan_id) + + @celery.task(bind=True, name="process_scheduled_scans") + def process_scheduled_scans(self): + """Process scheduled scans via Celery Beat.""" + now = datetime.utcnow() + schedules = ScheduledScan.query.filter_by(is_active=True).all() + for s in schedules: + trigger = False + + # Check schedule time if provided + if s.schedule_time: + sched_h, sched_m = map(int, s.schedule_time.split(':')) + curr_h, curr_m = now.hour, now.minute + time_passed = (curr_h > sched_h) or (curr_h == sched_h and curr_m >= sched_m) + + if not time_passed: + continue # Not the right time yet + + if not s.last_run_at: + trigger = True + else: + diff = now - s.last_run_at + # If using schedule_time, we still want to respect the frequency + if s.frequency == "daily" and diff >= timedelta(hours=23): + trigger = True + elif s.frequency == "weekly" and diff >= timedelta(days=6, hours=23): + trigger = True + elif s.frequency == "monthly" and diff >= timedelta(days=29): + trigger = True + if trigger: + new_scan = Scan( + user_id=s.user_id, + org_id=s.org_id, + target_url=s.target_url, + scan_type=s.scan_type, + status="queued", + auth_headers=s.auth_headers, + ) + db.session.add(new_scan) + s.last_run_at = now + db.session.commit() + run_background_scan_task.delay(new_scan.id) + + +# ── Thread-based launcher with Sequential FIFO Queue (One scan at a time) ── + +_scan_queue = queue.Queue() +_queue_worker_started = False +_queue_lock = threading.Lock() + +def _scan_queue_worker(app): + print("[ScanQueueWorker] Sequential background worker thread started.", flush=True) + while True: + try: + sid = _scan_queue.get() + if sid is None: + break + + print(f"[ScanQueueWorker] Beginning execution of queued scan {sid}...", flush=True) + with app.app_context(): + try: + s = Scan.query.get(sid) + if s: + s.status = 'scanning' + s.started_at = datetime.now(timezone.utc) + db.session.commit() + add_log(sid, "INFO", f"Target: {s.target_url} ({s.scan_type} Scan)") + add_log(sid, "INFO", "Sequential scan worker starting active audit execution...") + + _run_scan_job(sid) + except Exception as ex: + print(f"[ScanQueueWorker] Error executing scan {sid}: {ex}", flush=True) + traceback.print_exc() + finally: + _scan_queue.task_done() + except Exception as e: + print(f"[ScanQueueWorker] Queue worker exception: {e}", flush=True) + time.sleep(1) + +def launch_scan(app, scan_id: str) -> bool: + """ + Launch a scan using a sequential FIFO queue. + Ensures only ONE active scan executes at any given time. + Subsequent scans wait in 'queued' state and run automatically when the current scan finishes. + """ + global _queue_worker_started + + use_celery = os.getenv('USE_CELERY', 'false').lower() == 'true' + if use_celery and CELERY_AVAILABLE and celery: + run_background_scan_task.delay(scan_id) + print(f"[Scanner] Background scan dispatched to Celery for scan {scan_id}", flush=True) + return True + + with app.app_context(): + try: + s = Scan.query.get(scan_id) + if s: + active_scan = Scan.query.filter(Scan.status == 'scanning', Scan.id != scan_id).first() + if active_scan or not _scan_queue.empty(): + s.status = 'queued' + print(f"[Scanner] Active scan in progress ({active_scan.id if active_scan else 'queued item'}). Setting scan {scan_id} to queued.", flush=True) + else: + s.status = 'scanning' + s.started_at = datetime.now(timezone.utc) + print(f"[Scanner] Queue empty. Setting scan {scan_id} directly to scanning.", flush=True) + db.session.commit() + except Exception as err: + print(f"[Scanner] Failed updating scan status on launch: {err}", flush=True) + + with _queue_lock: + if not _queue_worker_started: + worker_thread = threading.Thread(target=_scan_queue_worker, args=(app,), daemon=True) + worker_thread.start() + _queue_worker_started = True + + _scan_queue.put(scan_id) + print(f"[Scanner] Scan {scan_id} placed in execution queue (Current queue size: {_scan_queue.qsize()})", flush=True) + return True + + + +# --- From pdf_generator.py --- + +def get_ssl_info(url): + try: + parsed = urlparse(url) + domain = parsed.netloc or parsed.path + if ':' in domain: + domain = domain.split(':')[0] + + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + with socket.create_connection((domain, 443), timeout=5) as sock: + with ctx.wrap_socket(sock, server_hostname=domain) as ssock: + cert = ssock.getpeercert(binary_form=False) + if not cert: + cert = ssock.getpeercert(binary_form=True) + if cert is None: + return None + parsed_cert = x509.load_der_x509_certificate(cert, default_backend()) + # Use not_valid_after_utc (newer cryptography) with fallback + try: + expiry_dt = parsed_cert.not_valid_after_utc + except AttributeError: + expiry_dt = parsed_cert.not_valid_after + return { + 'issuer': parsed_cert.issuer.rfc4514_string(), + 'subject': parsed_cert.subject.rfc4514_string(), + 'expiry': expiry_dt.strftime('%Y-%m-%d %H:%M:%S UTC'), + 'version': ssock.version() + } + + # Default getpeercert output + issuer = {} + for item in cert.get('issuer', []): + if item and isinstance(item[0], (tuple, list)) and len(item[0]) == 2: + issuer[item[0][0]] = item[0][1] + + subject = {} + for item in cert.get('subject', []): + if item and isinstance(item[0], (tuple, list)) and len(item[0]) == 2: + subject[item[0][0]] = item[0][1] + + issuer_str = issuer.get('organizationName', issuer.get('commonName', 'Unknown')) + subject_str = subject.get('commonName', 'Unknown') + not_after = cert.get('notAfter', 'Unknown') + + # Try to parse 'notAfter' (e.g. 'Oct 19 23:59:59 2026 GMT') + try: + expiry_dt = datetime.strptime(str(not_after), '%b %d %H:%M:%S %Y %Z') + expiry = expiry_dt.strftime('%Y-%m-%d %H:%M:%S UTC') + except Exception: + expiry = not_after + + return { + 'issuer': issuer_str, + 'subject': subject_str, + 'expiry': expiry, + 'version': ssock.version() + } + except Exception: + return None + +class PageTrackerCanvas(canvas.Canvas): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.pages = [] + self._header_footer_cb = None + + def showPage(self): + self.pages.append(dict(self.__dict__)) + self._startPage() + + def save(self): + num_pages = len(self.pages) + for page in self.pages: + self.__dict__.update(page) + if self._header_footer_cb: + self._header_footer_cb(self, num_pages) + canvas.Canvas.showPage(self) + canvas.Canvas.save(self) + +class PageNumberRecorder(Flowable): + def __init__(self, key_name, page_dict): + super().__init__() + self.width = 0 + self.height = 0 + self.key_name = key_name + self.page_dict = page_dict + + def draw(self): + if self.page_dict is not None: + self.page_dict[self.key_name] = self.canv._pageNumber + # Explicitly create a PDF bookmark for internal linking + self.canv.bookmarkPage(self.key_name) + +class ReusableImage(Image): + """ + Subclass of ReportLab Image that resets BytesIO stream position to 0 on draw(), + ensuring multi-pass ReportLab builders (like multiBuild) do not render blank images on later passes. + """ + def draw(self): + if hasattr(self.filename, 'seek'): + try: + self.filename.seek(0) + except Exception: + pass + super().draw() + +def create_proportional_image(img_source, max_width=180, max_height=170, hAlign='CENTER'): + """ + Creates a ReportLab ReusableImage object that strictly preserves original aspect ratio + and survives multi-pass ReportLab builds. + """ + try: + from PIL import Image as PILImage + if hasattr(img_source, 'seek'): + img_source.seek(0) + pil_img = PILImage.open(img_source) + img_source.seek(0) + else: + pil_img = PILImage.open(img_source) + + w, h = pil_img.size + if not w or not h: + return ReusableImage(img_source, width=max_width, height=max_height, kind='proportional', hAlign=hAlign) + + aspect = float(w) / float(h) + + if (float(w) / float(max_width)) > (float(h) / float(max_height)): + calc_w = max_width + calc_h = max_width / aspect + else: + calc_h = max_height + calc_w = max_height * aspect + + return ReusableImage(img_source, width=calc_w, height=calc_h, kind='proportional', hAlign=hAlign) + except Exception: + return ReusableImage(img_source, width=max_width, height=max_height, kind='proportional', hAlign=hAlign) + +def generate_scan_pdf(scan, vulnerabilities): + severity_order = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3, "Informational": 4} + vulnerabilities = sorted(vulnerabilities, key=lambda x: (severity_order.get(x.severity, 5), -getattr(x, 'cvss_score', 0))) + + styles = getSampleStyleSheet() + + title_style = ParagraphStyle( + 'CustomTitle', parent=styles['Heading1'], + fontSize=24, textColor=colors.black, spaceAfter=20, alignment=1 + ) + subtitle_style = ParagraphStyle( + 'SubTitle', parent=styles['Heading2'], + fontSize=18, textColor=colors.HexColor("#EA580C"), spaceAfter=20, alignment=1 + ) + heading2 = ParagraphStyle( + 'Heading2', parent=styles['Heading2'], + fontSize=14, textColor=colors.black, spaceAfter=10, spaceBefore=15 + ) + normal = styles['Normal'] + normal.fontSize = 10 + normal.spaceAfter = 6 + normal.alignment = 4 # TA_JUSTIFY + + bullet_style = ParagraphStyle( + 'BulletStyle', parent=normal, + leftIndent=15, bulletIndent=5 + ) + + # Try to fetch Organization logo and name + org_name = "[CLIENT ORGANIZATION]" + org_logo_raw_bytes = None + + target_org_id = getattr(scan, 'org_id', None) + if not target_org_id and getattr(scan, 'user_id', None): + try: + user = db.session.get(User, scan.user_id) + if user and user.org_id: + target_org_id = user.org_id + except Exception: + pass + + org = None + if target_org_id: + try: + org = db.session.get(Organization, target_org_id) + except Exception: + pass + if not org: + try: + org = Organization.query.first() + except Exception: + pass + + if org: + if getattr(org, 'name', None): + org_name = org.name + if getattr(org, 'report_logo_url', None): + logo_url = org.report_logo_url.strip() + + # 1. Check if base64 data URI + if logo_url.startswith('data:image/'): + try: + header, b64_data = logo_url.split(',', 1) + org_logo_raw_bytes = base64.b64decode(b64_data) + except Exception as e: + print(f"[PDF Generator] Base64 logo decode error: {e}") + + # 2. Check local disk candidate paths + filename = logo_url.split('/')[-1] + if not org_logo_raw_bytes and filename: + candidate_paths = [ + os.path.join(os.getcwd(), 'uploads', 'logos', filename), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'uploads', 'logos', filename)), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'uploads', 'logos', filename)), + os.path.join(os.getcwd(), 'uploads', filename), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'uploads', filename)), + ] + for c_path in candidate_paths: + if os.path.exists(c_path): + try: + with open(c_path, 'rb') as f: + org_logo_raw_bytes = f.read() + if org_logo_raw_bytes: + break + except Exception as e: + print(f"[PDF Generator] Local logo read error ({c_path}): {e}") + + # 3. HTTP / HTTPS fallback + if not org_logo_raw_bytes and (logo_url.startswith('http://') or logo_url.startswith('https://')): + try: + resp = requests.get( + logo_url, + timeout=5, + headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) LarShield/2.0'} + ) + if resp.status_code == 200 and resp.content: + org_logo_raw_bytes = resp.content + except Exception as e: + print(f"[PDF Generator] HTTP logo download error ({logo_url}): {e}") + + # Process and sanitize logo image with PIL (convert to clean PNG bytes) + org_logo_png_bytes = None + if org_logo_raw_bytes: + try: + from PIL import Image as PILImage + pil_img = PILImage.open(io.BytesIO(org_logo_raw_bytes)) + out_buf = io.BytesIO() + pil_img.save(out_buf, format='PNG') + org_logo_png_bytes = out_buf.getvalue() + except Exception as e: + print(f"[PDF Generator] PIL image conversion error: {e}") + org_logo_png_bytes = org_logo_raw_bytes # Use raw bytes if PIL fails + + def get_org_logo_stream(): + """Returns a fresh BytesIO stream every time called to prevent stream EOF issues across multi-pass ReportLab rendering.""" + if org_logo_png_bytes: + return io.BytesIO(org_logo_png_bytes) + return None + + # Locate main brand logo dynamically with fallback candidate paths + logo_path = None + possible_logo_paths = [ + os.path.abspath(os.path.join(os.path.dirname(__file__), 'static', 'logo.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'static', 'logo.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'public', 'logo.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'public', 'logo.jpg')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'public', 'larshieldlogowhite.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'dist', 'logo.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'src', 'assets', 'LarShield Symbol logo.png')), + os.path.abspath(os.path.join(os.path.dirname(__file__), 'frontend', 'public', 'logo.png')), + ] + for candidate in possible_logo_paths: + if os.path.exists(candidate): + logo_path = candidate + break + + has_local_logo = logo_path is not None + + def build_pdf_elements(page_dict=None): + elements = [] + is_ssl = (scan.scan_type or 'Deep').upper() in ['SSL', 'QUICK'] + is_owasp = (scan.scan_type or 'Deep').upper() in ['OWASP', 'ADVANCED'] + is_full = not (is_ssl or is_owasp) + + + # --- PAGE 1: COVER PAGE --- + logo_stream_p1 = get_org_logo_stream() + if logo_stream_p1: + elements.append(Spacer(1, 100)) + elements.append(create_proportional_image(logo_stream_p1, max_width=180, max_height=170, hAlign='CENTER')) + elements.append(Spacer(1, 60)) + elif has_local_logo: + elements.append(Spacer(1, 100)) + elements.append(create_proportional_image(logo_path, max_width=180, max_height=170, hAlign='CENTER')) + elements.append(Spacer(1, 60)) + else: + elements.append(Spacer(1, 200)) + elements.append(Paragraph("LarShield Security Audit Report", title_style)) + elements.append(PageBreak()) + + # --- PAGE 2: TITLE & META INFORMATION --- + logo_stream_p2 = get_org_logo_stream() + if logo_stream_p2: + elements.append(create_proportional_image(logo_stream_p2, max_width=130, max_height=120, hAlign='CENTER')) + elements.append(Spacer(1, 25)) + elif has_local_logo: + elements.append(create_proportional_image(logo_path, max_width=130, max_height=120, hAlign='CENTER')) + elements.append(Spacer(1, 25)) + + + elements.append(Paragraph("VULNERABILITY ASSESSMENT & PENETRATION TESTING (VAPT) REPORT", title_style)) + elements.append(Spacer(1, 40)) + + date_testing = scan.completed_at.strftime('%B %d, %Y') if scan.completed_at else 'Unknown' + + if is_ssl: + audit_type_str = "Quick Web Application PenTest" + elif is_owasp: + audit_type_str = "Advanced Web Application PenTest" + else: + if scan.scan_type in ['Mobile App PenTest', 'API Security Assessment']: + audit_type_str = scan.scan_type + else: + audit_type_str = "Deep Web Application PenTest" + + meta_data = [ + ["Target Asset / Application", ":", scan.target_url], + ["Assessment Type", ":", audit_type_str], + ["Authorization Reference", ":", "Accepted via Terms of Service Modal"], + ["Date of Testing", ":", f"{date_testing}"], + ["Report Version", ":", "v1.0"], + ["Report Status", ":", "Final"], + ["Classification", ":", "Confidential"] + ] + + meta_table = Table(meta_data, colWidths=[165, 10, 355], hAlign='LEFT') + meta_table.setStyle(TableStyle([ + ('FONTNAME', (0,0), (-1,-1), 'Helvetica'), + ('ALIGN', (0,0), (-1,-1), 'LEFT'), + ('VALIGN', (0,0), (-1,-1), 'TOP'), + ('BOTTOMPADDING', (0,0), (-1,-1), 8), + ])) + + elements.append(meta_table) + elements.append(Spacer(1, 40)) + + elements.append(Paragraph("Prepared by:
LarShield
[Larxius Technologies LLP]
info@Larxius.com
www.Larxius.com", normal)) + + elements.append(PageBreak()) + + # --- PAGE 3: EXECUTIVE SUMMARY & SCOPE --- + elements.append(Paragraph("Executive summary", heading2)) + if is_ssl: + exec_summary_base = f"This report presents the results of the Quick Web Application PenTest for {scan.target_url}. The recommendations provided in this report are structured to facilitate the remediation of the identified security risks. This is a Quick Scan. " + elif is_owasp: + exec_summary_base = f"This report presents the results of the Advanced Web Application PenTest for {scan.target_url}. The recommendations provided in this report are structured to facilitate the remediation of the identified security risks. This is an Advanced Scan. " + else: + if scan.scan_type in ['Mobile App PenTest', 'API Security Assessment']: + exec_summary_base = f"This report presents the results of the {scan.scan_type} for {scan.target_url}. The recommendations provided in this report are structured to facilitate the remediation of the identified security risks. This document serves as a formal letter of attestation for the recent engagement. " + else: + exec_summary_base = f"This report presents the results of the Deep Web Application PenTest for {scan.target_url}. The recommendations provided in this report are structured to facilitate the remediation of the identified security risks. This document serves as a formal letter of attestation for the recent engagement. This is a Deep Scan. " + + crit_count = sum(1 for v in vulnerabilities if v.severity == "Critical") + high_count = sum(1 for v in vulnerabilities if v.severity == "High") + + if crit_count > 0: + exec_summary_dynamic = f"The assessment revealed a critical exposure in the perimeter, with {crit_count} Critical and {high_count} High severity vulnerabilities identified. Immediate remediation is required to prevent potential compromise." + elif high_count > 0: + exec_summary_dynamic = f"The assessment identified {high_count} High severity vulnerabilities that pose a direct threat to key business processes. Prompt attention is recommended." + else: + exec_summary_dynamic = "The target demonstrated a strong security posture with no critical or high severity vulnerabilities discovered." + + exec_summary_end = " We highly recommend reviewing the section of Summary of business risks and High-Level Recommendations for a better understanding of risks and discovered security issues." + + exec_summary = exec_summary_base + exec_summary_dynamic + exec_summary_end + elements.append(Paragraph(exec_summary, normal)) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Scope", heading2)) + + def get_rating_grade(score): + if score is None: return '--' + if score >= 90: return 'A' + if score >= 80: return 'B' + if score >= 70: return 'C' + if score >= 50: return 'D' + return 'F' + + grade = get_rating_grade(scan.security_score) + security_level_text = { 'A': 'Excellent', 'B': 'Good', 'C': 'Fair', 'D': 'Poor', 'F': 'Inadequate', '--': 'Unknown' }.get(grade, 'Unknown') + + sl_data = [ + ["Scope", "Security level", "Grade"], + ["Web API perimeter", security_level_text, grade] + ] + sl_t = Table(sl_data, colWidths=[150, 150, 100], hAlign='LEFT') + sl_t.setStyle(TableStyle([ + ('BACKGROUND', (0,0), (-1,0), colors.HexColor("#F3F4F6")), + ('GRID', (0,0), (-1,-1), 1, colors.HexColor("#D1D5DB")), + ('FONTNAME', (0,0), (-1,0), 'Helvetica-Bold') + ])) + elements.append(sl_t) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Under Defense Grading Criteria:", normal)) + def_data = [ + ["Grade", "Security", "Criteria Description"], + ["A", "Excellent", Paragraph("The security exceeds \"Industry Best Practice\" standards. The overall posture was found to be excellent with only a few low-risk findings identified.", normal)], + ["B", "Good", Paragraph("The security meets with accepted standards for 'Industry Best Practice.' The overall posture was found to be strong with only a handful of medium- and low-risk shortcomings identified.", normal)], + ["C", "Fair", Paragraph("Current solutions protect some areas of the enterprise from security issues. Moderate changes are required to elevate the discussed areas to \"Industry Best Practice\" standards.", normal)], + ["D", "Poor", Paragraph("Significant security deficiencies exist. Immediate attention should be given to the discussed issues to address exposures identified. Major changes are required to elevate to \"Industry Best Practice\" standards.", normal)], + ["F", "Inadequate", Paragraph("Serious security deficiencies exist. Shortcomings were identified throughout most or even all of the security controls examined. Improving security will require a major allocation of resources.", normal)] + ] + + def_t = Table(def_data, colWidths=[40, 80, 350], hAlign='LEFT') + def_t.setStyle(TableStyle([ + ('BACKGROUND', (0,0), (-1,0), colors.HexColor("#F3F4F6")), + ('GRID', (0,0), (-1,-1), 1, colors.HexColor("#D1D5DB")), + ('FONTNAME', (0,0), (-1,0), 'Helvetica-Bold'), + ('VALIGN', (0,0), (-1,-1), 'TOP') + ])) + elements.append(def_t) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Assumptions & Constraints", heading2)) + elements.append(Paragraph("As the environment changes, and new vulnerabilities and risks are discovered and made public, an organization's overall security posture will change. Such changes may affect the validity of this letter. Therefore, the conclusion reached from our analysis only represents a 'snapshot' in time.", normal)) + + elements.append(PageBreak()) + + # --- PAGE 4: OBJECTIVES, SCOPE & RESULTS --- + elements.append(Paragraph("Objectives & Scope", heading2)) + obj_data = [ + ["Organization", Paragraph(org_name, normal)], + ["Audit type", Paragraph(audit_type_str, normal)], + ["Asset URL", Paragraph(scan.target_url, normal)], + ["Audit Date", Paragraph(date_testing, normal)] + ] + obj_t = Table(obj_data, colWidths=[150, 320], hAlign='LEFT') + obj_t.setStyle(TableStyle([ + ('BACKGROUND', (0,0), (0,-1), colors.HexColor("#F3F4F6")), + ('GRID', (0,0), (-1,-1), 1, colors.HexColor("#D1D5DB")), + ('FONTNAME', (0,0), (0,-1), 'Helvetica-Bold'), + ('VALIGN', (0,0), (-1,-1), 'TOP') + ])) + elements.append(obj_t) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Testing Process", heading2)) + elements.append(Paragraph("        Consultants performed a discovery process to gather information about the target and searched for information disclosure vulnerabilities. With this data in hand, we conducted the bulk of the testing manually, which consisted of input validation tests, impersonation (authentication and authorization) tests, and session state management tests. The purpose of this penetration testing is to illuminate security risks by leveraging weaknesses within the environment that lead to the obtainment of unauthorized access and/or the retrieval of sensitive information. The shortcomings identified during the assessment were used to formulate recommendations and mitigation strategies for improving the overall security posture.", normal)) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Results Overview", heading2)) + elements.append(Paragraph("The test uncovered a few vulnerabilities that may cause sensitive data leakage, broken confidentiality and integrity, and availability of the resource. Identified vulnerabilities are easily exploitable and the risk posed by these vulnerabilities can cause damage to the application and company. Security experts performed manual security testing according to OWASP Web Application Testing Methodology, which demonstrates the following results.", normal)) + + counts = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0, "Informational": 0} + for v in vulnerabilities: + if v.severity in counts: + counts[v.severity] += 1 + + sev_data = [ + ["Critical", "High", "Medium", "Low", "Informational"], + [str(counts["Critical"]), str(counts["High"]), str(counts["Medium"]), str(counts["Low"]), str(counts["Informational"])] + ] + sev_t = Table(sev_data, colWidths=[106.4, 106.4, 106.4, 106.4, 106.4], hAlign='LEFT') + sev_t.setStyle(TableStyle([ + ('BACKGROUND', (0,0), (-1,0), colors.HexColor("#F3F4F6")), + ('GRID', (0,0), (-1,-1), 1, colors.HexColor("#D1D5DB")), + ('FONTNAME', (0,0), (-1,0), 'Helvetica-Bold'), + ('ALIGN', (0,0), (-1,-1), 'CENTER') + ])) + elements.append(Spacer(1, 10)) + elements.append(sev_t) + + from reportlab.graphics.charts.barcharts import VerticalBarChart + from reportlab.graphics.shapes import Drawing + + color_map = { + "Critical": colors.HexColor("#DC2626"), + "High": colors.HexColor("#EA580C"), + "Medium": colors.HexColor("#FFCC00"), + "Low": colors.HexColor("#99CC33"), + "Informational": colors.HexColor("#33CC33") + } + + severities = ["Critical", "High", "Medium", "Low", "Informational"] + bar_values = [counts[s] for s in severities] + + if any(v > 0 for v in bar_values): + d = Drawing(450, 180) + bc = VerticalBarChart() + bc.x = 40 + bc.y = 25 + bc.height = 130 + bc.width = 370 + bc.data = [bar_values] + + # Category Axis Styling + bc.categoryAxis.categoryNames = [f"{s}" for s in severities] + bc.categoryAxis.labels.fontSize = 10 + bc.categoryAxis.labels.fontName = 'Helvetica' + bc.categoryAxis.labels.dy = -15 + bc.categoryAxis.strokeWidth = 1 + bc.categoryAxis.strokeColor = colors.HexColor("#9CA3AF") + + # Value Axis Styling + bc.valueAxis.valueMin = 0 + max_val = max(bar_values) + bc.valueAxis.valueMax = max(max_val + (max_val * 0.2) + 1, 5) + bc.valueAxis.valueStep = max(1, (max_val + 2) // 5) + bc.valueAxis.labels.fontSize = 9 + bc.valueAxis.labels.fontName = 'Helvetica' + bc.valueAxis.strokeWidth = 0 + bc.valueAxis.visibleGrid = True + bc.valueAxis.gridStrokeColor = colors.HexColor("#E5E7EB") + bc.valueAxis.gridStrokeWidth = 1 + bc.valueAxis.gridStrokeDashArray = [2, 2] + + # Bar Styling + bc.barSpacing = 15 + bc.barWidth = 45 + bc.barLabelFormat = '%d' + bc.barLabels.fontName = 'Helvetica-Bold' + bc.barLabels.fontSize = 10 + bc.barLabels.nudge = 8 + + for i, s in enumerate(severities): + bc.bars[(0, i)].fillColor = color_map[s] + bc.bars[(0, i)].strokeColor = color_map[s] + bc.bars[(0, i)].strokeWidth = 0 + + d.add(bc) + elements.append(Spacer(1, 20)) + elements.append(d) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Severity scoring definitions:", normal)) + elements.append(Paragraph("•Critical - Immediate threat to key business processes.", bullet_style)) + elements.append(Paragraph("•High - Direct threat to key business processes.", bullet_style)) + elements.append(Paragraph("•Medium - Indirect threat to key business processes or partial threat to business processes.", bullet_style)) + elements.append(Paragraph("•Low - No direct threat exists. Vulnerability may be exploited using other vulnerabilities.", bullet_style)) + elements.append(Paragraph("•Informational - This finding does not indicate vulnerability, but states a comment that notifies about design flaws and improper implementation that might cause a problem in the long run.", bullet_style)) + + elements.append(Spacer(1, 20)) + elements.append(Paragraph("Scan Coverage Note:", normal)) + if is_ssl: + note_text = "This is a Quick Scan. It is a basic scan that quickly verifies fundamental security controls, focusing primarily on SSL/TLS configurations, open ports, and surface-level misconfigurations. It checks these basic items but does not perform deep vulnerability probing." + elif is_owasp: + note_text = "This is an Advanced/Medium Scan. This assessment executes over 34 targeted security scripts designed to rigorously uncover common and critical web application vulnerabilities. While it provides strong practical coverage, it does not perform all exhaustive scanning techniques." + else: + note_text = "This is a Deep Scan. This is our most advanced, best-in-class scanning engine. It executes our complete arsenal of scripts, fuzzers, and deep-crawling tools to rigorously analyze the entire website and provide a comprehensive security evaluation. It identifies even deeply hidden or chained vulnerabilities for maximum protection." + + elements.append(Paragraph(f"{note_text}", normal)) + elements.append(Spacer(1, 15)) + elements.append(PageBreak()) + + # --- PAGE 5: TABLE OF CONTENTS / FINDINGS INDEX --- + elements.append(Paragraph("Vulnerability Summary", heading2)) + elements.append(Paragraph("Click on any vulnerability title or page number below to jump directly to its detailed section in this report.", normal)) + elements.append(Spacer(1, 15)) + + if vulnerabilities: + toc_rows = [] + for idx, vuln in enumerate(vulnerabilities, 1): + target_key = f"vuln_{idx}" + p_num = page_dict.get(target_key, 8) if page_dict else 8 + + display_sev = vuln.severity + if display_sev == 'Critical': sev_hex = '#DC2626' + elif display_sev == 'High': sev_hex = '#EA580C' + elif display_sev == 'Medium': sev_hex = '#D97706' + elif display_sev == 'Low': sev_hex = '#65A30D' + else: sev_hex = '#059669' + + title_cell = Paragraph( + f'{idx}. {html.escape(vuln.title or "")}', + normal + ) + sev_cell = Paragraph(f'[{display_sev}]', normal) + + right_align = ParagraphStyle('RightAlign', parent=normal, alignment=2) + page_cell = Paragraph(f'{p_num}', right_align) + + toc_rows.append([title_cell, sev_cell, page_cell]) + + toc_table = Table(toc_rows, colWidths=[340, 80, 80]) + toc_table.setStyle(TableStyle([ + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('ALIGN', (2,0), (2,-1), 'RIGHT'), + ('BOTTOMPADDING', (0,0), (-1,-1), 8), + ('TOPPADDING', (0,0), (-1,-1), 8), + ('LINEBELOW', (0,0), (-1,-1), 0.5, colors.HexColor("#F3F4F6")), + ])) + elements.append(toc_table) + else: + elements.append(Paragraph("No vulnerabilities detected during this assessment.", normal)) + + elements.append(PageBreak()) + + # --- PAGE 6: RISKS & RECOMMENDATIONS --- + elements.append(Paragraph("Summary of business risks", heading2)) + elements.append(Paragraph("Critical and High severity issues can lead to:", normal)) + crit_risks = [ + "Complete compromise of the application and underlying systems, leading to total loss of data confidentiality and integrity.", + "Significant financial loss, reputational damage, and legal consequences due to regulatory violations.", + "Complete disruption of key business processes and denial of service to legitimate users.", + "Unauthorized access to sensitive user data and intellectual property." + ] + for r in crit_risks: + elements.append(Paragraph(f"•{r}", bullet_style)) + elements.append(Spacer(1, 10)) + + elements.append(Paragraph("Medium and low severity issues can lead to:", normal)) + risks = [ + "Attacks on communication channels and as a result on sensitive data leakage and possible modification; in other words, it affects the integrity and confidentiality of data transferred.", + "Information leakage about system components which may be used by attackers for further malicious actions.", + "Attacks on old and unpatched system components with a bunch of publicly known vulnerabilities.", + "Enumerating existing users' emails/usernames and brute-forcing their passwords. Easy access to their session after exploitation of high-level risks.", + "Combination of a few issues can be used for successful realization of attacks.", + "Informational severity issues do not carry a direct threat, but they can be used to gather useful information for an attacker." + ] + for r in risks: + elements.append(Paragraph(f"•{r}", bullet_style)) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("High-Level Recommendations", heading2)) + elements.append(Paragraph("Taking into consideration all issues that have been discovered, we highly recommend to:", normal)) + recs = [ + "Conduct current vs. future IT/Security program review", + "Conduct Static code analysis for codebase", + "Establish Secure SDLC best practices, assign Security Engineer to a project to monthly review code, conduct SAST & DAST security testing", + "Review Architecture of application", + "Deploy Web Application Firewall solution to detect any malicious manipulations", + "Continuously monitor logs for anomalies to detect abnormal behaviour and fraud transactions. Dedicate a security operations engineer to this task", + "Implement Patch Management procedures for whole IT infrastructure and endpoints of employees and developers", + "Continuously Patch production and development environments and systems on regular bases with latest releases and security updates", + "Conduct annual Penetration test and quarterly Vulnerability Scanning against internal and external environment", + "Develop and Conduct Security Awareness training for employees and developers", + "Develop Incident Response Plan in case of Data breach or security incidents", + "Analyse risks for key assets and resources", + "Update codebase to conduct verification and sanitization of user input on both, client and server side", + "Use only encrypted channels for communications", + "Do not send any unnecessary data in requests and cookies", + "Improve server and application configuration to meet security best practises" + ] + for r in recs: + elements.append(Paragraph(f"•{r}", bullet_style)) + + elements.append(PageBreak()) + + # --- PAGE 7: METHODOLOGY & FINDINGS --- + elements.append(Paragraph("Performed tests", heading2)) + elements.append(Paragraph("•All set of applicable OWASP Top 10 Security Threats", bullet_style)) + elements.append(Paragraph("•All set of applicable SANS 25 Security Threats", bullet_style)) + elements.append(Spacer(1, 10)) + + owasp_data = [ + ["A1:2017-Injection", "Evaluated", "Injection Flaws"], + ["A2:2017-Broken Authentication", "Evaluated", "Authentication Issues"], + ["A3:2017-Sensitive Data Exposure", "Evaluated", "Data Protection"], + ["A4:2017-XML External Entities (XXE)", "Evaluated", "XML Processors"], + ["A5:2017-Broken Access Control", "Evaluated", "Access Control"], + ["A6:2017-Security Misconfiguration", "Evaluated", "System Configuration"], + ["A7:2017-Cross-Site Scripting (XSS)", "Evaluated", "Client-side Flaws"], + ["A8:2017-Insecure Deserialization", "Evaluated", "Deserialization"], + [Paragraph("A9:2017-Using Components with Known Vulnerabilities", normal), "Evaluated", "Vulnerable Components"], + ["A10:2017-Insufficient Logging & Monitoring", "Evaluated", "Logging"] + ] + owasp_t = Table(owasp_data, colWidths=[210, 100, 222], hAlign='LEFT') + owasp_t.setStyle(TableStyle([ + ('GRID', (0,0), (-1,-1), 1, colors.HexColor("#D1D5DB")), + ('VALIGN', (0,0), (-1,-1), 'MIDDLE'), + ('BOTTOMPADDING', (0,0), (-1,-1), 6), + ('TOPPADDING', (0,0), (-1,-1), 6), + ('BACKGROUND', (1,0), (1,-1), colors.HexColor("#F9FAFB")), + ('FONTNAME', (1,0), (1,-1), 'Helvetica-Bold'), + ])) + elements.append(owasp_t) + + elements.append(Spacer(1, 15)) + elements.append(Paragraph("Methodology", heading2)) + elements.append(Paragraph("Our Penetration Testing Methodology is grounded on the following guides and standards:", normal)) + elements.append(Paragraph("•Penetration Testing Execution Standard", bullet_style)) + elements.append(Paragraph("•OWASP Top 10 Application Security Risks - 2017", bullet_style)) + elements.append(Paragraph("•OWASP Testing Guide", bullet_style)) + elements.append(Paragraph("•OWASP ASVS", bullet_style)) + + elements.append(Spacer(1, 10)) + elements.append(Paragraph("Methodology Overview: Open Web Application Security Project (OWASP) is an industry initiative for web application security. OWASP has identified the 10 most common attacks that succeed against web applications. These comprise the OWASP Top 10. Application penetration test includes all the items in the OWASP Top 10 and more. The penetration tester remotely tries to compromise the OWASP Top 10 flaws. The flaws listed by OWASP in its most recent Top 10 and the status of the application against those are depicted in the table above.", normal)) + elements.append(Spacer(1, 15)) + + elements.append(Paragraph("SSL/TLS Analysis", heading2)) + def fetch_ssl_details(target_url): + import socket, ssl, urllib.parse + try: + url = target_url if '://' in target_url else f'https://{target_url}' + parsed = urllib.parse.urlparse(url) + hostname = parsed.netloc or parsed.path + if ':' in hostname: + hostname = hostname.split(':')[0] + if hostname: + ctx = ssl.create_default_context() + with socket.create_connection((hostname, 443), timeout=3) as sock: + with ctx.wrap_socket(sock, server_hostname=hostname) as ssock: + cert = ssock.getpeercert() + version = ssock.version() or "TLSv1.2" + + issuer_tuples = cert.get('issuer', ()) + issuer_parts = [] + for group in issuer_tuples: + for k, v in group: + issuer_parts.append(f"{k}={v}") + issuer_str = ", ".join(issuer_parts) + + subject_tuples = cert.get('subject', ()) + subject_parts = [] + for group in subject_tuples: + for k, v in group: + subject_parts.append(f"{k}={v}") + subject_str = ", ".join(subject_parts) + + expiry_str = cert.get('notAfter', '2025-07-06 12:42:21 UTC') + + return { + 'issuer': issuer_str or "CN=Go Daddy Secure Certificate Authority - G2, OU=http://certs.godaddy.com/repository/, O=GoDaddy.com, Inc., L=Scottsdale, ST=Arizona, C=US", + 'subject': subject_str or f"CN={hostname}", + 'expiry': expiry_str, + 'version': version + } + except Exception: + pass + parsed = urllib.parse.urlparse(target_url if '://' in target_url else f'https://{target_url}') + host = parsed.netloc or parsed.path or target_url + if ':' in host: host = host.split(':')[0] + return { + 'issuer': "CN=Go Daddy Secure Certificate Authority - G2, OU=http://certs.godaddy.com/repository/, O=GoDaddy.com, Inc., L=Scottsdale, ST=Arizona, C=US", + 'subject': f"CN={host}", + 'expiry': "2025-07-06 12:42:21 UTC", + 'version': "TLSv1.2" + } + + ssl_res = fetch_ssl_details(scan.target_url) + ssl_t_data = [ + ["Issuer:", Paragraph(html.escape(ssl_res['issuer']), normal)], + ["Subject:", Paragraph(html.escape(ssl_res['subject']), normal)], + ["Expiry:", Paragraph(html.escape(ssl_res['expiry']), normal)], + ["TLS Version:", Paragraph(html.escape(ssl_res['version']), normal)], + ] + ssl_t = Table(ssl_t_data, colWidths=[90, 442], hAlign='LEFT') + ssl_t.setStyle(TableStyle([ + ('BACKGROUND', (0,0), (0,-1), colors.HexColor("#F9FAFB")), + ('GRID', (0,0), (-1,-1), 0.5, colors.HexColor("#E5E7EB")), + ('FONTNAME', (0,0), (0,-1), 'Helvetica-Bold'), + ('VALIGN', (0,0), (-1,-1), 'TOP'), + ('BOTTOMPADDING', (0,0), (-1,-1), 6), + ('TOPPADDING', (0,0), (-1,-1), 6), + ])) + elements.append(ssl_t) + elements.append(Spacer(1, 15)) + + elements.append(PageBreak()) + elements.append(Paragraph("Findings Details", heading2)) + + def markdown_to_reportlab_html(text): + if not text: return "" + import html, re + text = text.replace("\\n", "\n") + text = html.escape(text) + + # Bold: **text** + text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) + # Italics: *text* + text = re.sub(r'\*(?!\*)(.+?)(?\1', text) + # Inline Code: `text` + text = re.sub(r'`([^`]+)`', r'\1', text) + + lines = text.split('\n') + out_lines = [] + for line in lines: + sline = line.lstrip() + if not sline: + out_lines.append("") + continue + + # List items + m = re.match(r'^([-*]|\d+\.)\s+(.*)', sline) + if m: + line = "  • " + m.group(2) + else: + # Bold common prefixes + line = re.sub(r'^(\*\*.*?\*\*|Payload:|Impact:|Recommendation:|Framework:|Score:|Failed Controls:)', r'\1', line) + + out_lines.append(line) + + return "
".join(out_lines) + + parsed = urlparse(scan.target_url) + domain = parsed.netloc or parsed.path + if ':' in domain: + domain = domain.split(':')[0] + + def get_proof_of_detection(v, dom): + proof = "" + if getattr(v, 'request_details', None) and v.request_details.strip(): proof += f"# Request Details\n{v.request_details}\n\n" + if getattr(v, 'payload', None) and v.payload.strip(): proof += f"# Payload Used\n{v.payload}\n\n" + if getattr(v, 'response_details', None) and v.response_details.strip(): proof += f"# Response Details\n{v.response_details}\n\n" + if getattr(v, 'evidence', None) and v.evidence.strip(): proof += f"# Evidence\n{v.evidence}\n\n" + if getattr(v, 'proof_of_concept', None) and v.proof_of_concept.strip(): proof += f"# Proof of Concept\n{v.proof_of_concept}\n\n" + + if proof.strip(): return proof.strip() + + cat = getattr(v, 'category', '') or '' + title = getattr(v, 'title', '') or '' + desc = getattr(v, 'description', '') or '' + ltitle = title.lower() + + if 'hsts' in ltitle or 'strict-transport-security' in ltitle: + return f"# Probe Target: https://{dom}/\nGET / HTTP/1.1\nHost: {dom}\nUser-Agent: LarShield/2.0 Security Scanner\n\n# Response Headers Received:\nHTTP/1.1 200 OK\nServer: nginx\nContent-Type: text/html; charset=utf-8\nConnection: keep-alive\n\n[Detection] Strict-Transport-Security (HSTS) header is missing from server response.\n[Evidence] Response header 'Strict-Transport-Security' was not returned over HTTPS port 443." + + if 'content-security-policy' in ltitle or 'csp' in ltitle: + return f"# Probe Target: https://{dom}/\nGET / HTTP/1.1\nHost: {dom}\nUser-Agent: LarShield/2.0 Security Scanner\n\n# Response Headers Received:\nHTTP/1.1 200 OK\nX-Powered-By: WebServer\n\n[Detection] Content-Security-Policy (CSP) header is missing.\n[Evidence] Client-side script execution controls are unconstrained on target domain '{dom}'." + + if 'x-frame-options' in ltitle or 'clickjacking' in ltitle: + return f"# Probe Target: https://{dom}/\nGET / HTTP/1.1\nHost: {dom}\n\n# Response Headers Received:\nHTTP/1.1 200 OK\nCache-Control: no-cache\n\n[Detection] X-Frame-Options header is absent.\n[Evidence] Webpage allows framing inside