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
)
try:
from .extensions import db, celery, socketio, limiter
from .models import *
except ImportError:
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": 120, # raised: 60 was too tight for slow probes
"standard": 180,
"advanced": 180,
"deep": 600,
}
_module_timeout = MODULE_TIMEOUTS.get((scan_type or "standard").lower(), 180)
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": 12, # raised: more concurrency to finish faster
"standard": 10,
"advanced": 10,
"deep": 8,
"ssl": 4,
"port": 2,
}
_max_workers = MAX_WORKERS_PER_PHASE.get((scan_type or "advanced").lower(), 10)
# Wall-clock hard limits per scan type (seconds)
HARD_LIMITS = {
"quick": 900, # 15 min (raised from 5 min — quick scan has 13 modules)
"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
# ── Seed logs immediately so the /logs endpoint never returns the static fallback ──
# This ensures the frontend sees real log entries from the very first poll,
# even if the queue worker hasn't started executing the scan yet.
try:
with app.app_context():
s = Scan.query.get(scan_id)
if s:
add_log(scan_id, "INFO", f"LarShield WSS — {s.scan_type.upper()} SCAN QUEUED")
add_log(scan_id, "INFO", f"Target: {s.target_url}")
add_log(scan_id, "INFO", f"Scan ID: {scan_id}")
add_log(scan_id, "INFO", "Waiting for execution slot in the sequential scan queue...")
except Exception as _seed_err:
print(f"[Scanner] Log seed warning (non-fatal): {_seed_err}", flush=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