import argparse import subprocess import sys import os import datetime import html import glob import json import re from typing import List, Tuple, Dict from test_descriptions import get_description import requests from requests.exceptions import RequestException from urllib.parse import urlparse # LLM server configuration: override via env var LLM_SERVER_URL # Default changed to your requested host (uses the /ask endpoint) LLM_SERVER_URL = os.environ.get('LLM_SERVER_URL', 'http://ashy.tplinkdns.com:5005/ask') def query_llm(prompt: str, system_prompt: str = None, timeout: int = None) -> str: """Query the local LLM server (/ask) and return the assistant response text. Returns empty string on error or timeout. """ provider = os.environ.get('LLM_PROVIDER', '') payload = {'prompt': prompt} if provider: payload['provider'] = provider if system_prompt: payload['system_prompt'] = system_prompt # Write prompt to debug log (best-effort) try: os.makedirs(OUTPUTS_DIR, exist_ok=True) with open(os.path.join(OUTPUTS_DIR, 'llm_debug.log'), 'a', encoding='utf-8') as lf: lf.write(f"\n--- PROMPT ({datetime.datetime.utcnow().isoformat()}) ---\n") lf.write(prompt[:10000] + "\n") except Exception: pass try: llm_timeout = timeout or int(os.environ.get('LLM_TIMEOUT', '30')) max_attempts = int(os.environ.get('LLM_RETRIES', '2')) resp = None for attempt in range(1, max_attempts + 1): try: resp = requests.post(LLM_SERVER_URL, json=payload, timeout=llm_timeout) if resp.status_code == 200: break except RequestException as ex: if attempt < max_attempts: time_to_wait = 0.3 * attempt try: import time time.sleep(time_to_wait) except Exception: pass else: raise if not resp or resp.status_code != 200: return '' data = resp.json() resp_text = (data.get('response') or '').strip() # append raw response to debug log try: with open(os.path.join(OUTPUTS_DIR, 'llm_debug.log'), 'a', encoding='utf-8') as lf: lf.write(f"--- RAW RESPONSE ({datetime.datetime.utcnow().isoformat()}) ---\n") lf.write(resp_text[:20000] + "\n") except Exception: pass # Try to parse assistant reply as JSON per our prompt guidance try: parsed = json.loads(resp_text) out_parts = [] if isinstance(parsed, dict): if 'summary' in parsed: out_parts.append(f"Summary: {parsed.get('summary','')}") if 'remediation' in parsed: rem = parsed.get('remediation') if isinstance(rem, list): out_parts.append("Remediation: " + "; ".join(rem)) else: out_parts.append("Remediation: " + str(rem)) if 'notes' in parsed: out_parts.append("Notes: " + str(parsed.get('notes',''))) return "\n".join([p for p in out_parts if p]) return resp_text except Exception: return resp_text except RequestException: return '' SCRIPTS_TO_RUN = [ "tech_fingerprinter.py", "header_checker.py", "robots_txt.py", "dns_whois_nslookup_checker.py", "cookie_checker.py", "spf_dmarc_checker.py", "waf_detector.py", "dependency_scanner.py", "sensitive_data_scanner.py", "sitemap_parser.py", "link_checker.py", "clickjacking_checker.py", "dirbust_scanner.py", "admin_finder.py", "cors_checker.py", "http_methods_checker.py" ] # Add additional checks mapping to OWASP categories SCRIPTS_TO_RUN += [ "tls_checker.py", "integrity_checker.py", "config_checker.py", "logging_checker.py", "auth_checker.py", ] PROJECT_DIR = os.path.dirname(__file__) OUTPUTS_DIR = os.path.join(PROJECT_DIR, "outputs") # Simple keyword -> severity mapping for automatic classification KEYWORD_SEVERITY = { 'VULNERABILITY': 'High', 'Potential sensitive data': 'High', 'OUTDATED': 'Medium', 'Missing Header': 'Medium', 'Broken Link Found': 'Low', 'Unreachable Link Found': 'Low', 'Could not fetch': 'Low', 'Error': 'Low' } def classify_line(line: str) -> Tuple[str, str]: """ Return (matched_keyword, severity) for a line if any keyword is found. """ for kw, sev in KEYWORD_SEVERITY.items(): if kw in line: return kw, sev return '', '' def run_script_and_capture(script: str, target: str) -> Tuple[str, List[dict], List[str], str]: """ Run a script and return raw output, parsed findings, suggestions, and an AI analysis string from the LLM server. Returns (output, findings_list, suggestions_list, ai_analysis) """ path = os.path.join(PROJECT_DIR, script) if not os.path.exists(path): return f"Script missing: {script}\n", [], [], '' try: proc = subprocess.run([sys.executable, path, target], capture_output=True, text=True, timeout=120) out = proc.stdout or "" err = proc.stderr or "" combined = out + ("\n--- STDERR ---\n" + err if err else "") # Remove ANSI escape sequences for clean HTML and parsing clean = re.sub(r'\x1b\[[0-9;]*m', '', combined) out = clean findings = [] suggestions = [] current_in_suggestions = False for raw_line in out.splitlines(): line = raw_line.strip() if not line: continue # Check if we're entering suggestions section if "SUGGESTIONS:" in line or "📋" in line: current_in_suggestions = True continue # Capture suggestions if current_in_suggestions and (line.startswith('1.') or line.startswith('2.') or line.startswith('3.') or line.startswith('4.') or line.startswith('5.') or line.startswith('-') or line[0].isdigit()): suggestions.append(line) continue elif current_in_suggestions and line.startswith('['): # End of suggestions section current_in_suggestions = False # Capture findings kw, sev = classify_line(line) if kw: findings.append({'line': line, 'keyword': kw, 'severity': sev, 'script': script}) # Prepare a concise prompt for the LLM summarising findings and suggestions try: # Include test name and a short reason/description before findings reason = '' try: reason = get_description(script) or '' except Exception: reason = '' prompt_lines = [f"AI SECURITY ANALYSIS", f"Test: {script}", f"Reason: {reason}", "\nFindings:"] if findings: for f in findings: prompt_lines.append(f"- {f.get('line','')}") else: prompt_lines.append("- No automated findings detected.") prompt_lines.append("\nSuggestions:") if suggestions: for s in suggestions[:10]: prompt_lines.append(f"- {s}") else: prompt_lines.append("- No explicit suggestions captured.") # Include a short excerpt of raw output to give context if out: excerpt = out[:2000] prompt_lines.append("\nOutput excerpt:") prompt_lines.append(excerpt) prompt_text = '\n'.join(prompt_lines) # Ask the LLM to respond with a compact JSON object for easy parsing system = ('You are a security analyst. Respond ONLY with a valid JSON object ' 'with keys: "summary" (concise but at least 2-3 sentences), ' '"remediation" (an array of remediation steps or a short string; each item should be 2-3 sentences), ' 'and optional "notes" (if present, provide 2-3 sentences). Return only the JSON object and no extra commentary. Keep the response machine-parseable.') ai_analysis = query_llm(prompt_text, system_prompt=system) except Exception: ai_analysis = '' return out, findings, suggestions, ai_analysis except Exception as e: return f"Error running {script}: {e}\n", [{'line': str(e), 'keyword': 'Error', 'severity': 'Low', 'script': script}], [], '' def find_screenshot_for_target(target: str) -> Tuple[str, str]: """ Return (inline_base64, file_path) tuple. If a screenshot JSON exists, return its base64 content as inline_base64 and empty file_path. Otherwise return (None, file_path) pointing to a PNG in outputs/ if found. """ domain = target.replace('http://','').replace('https://','').split('/')[0].replace(':','_') # Check JSON first json_path = os.path.join(OUTPUTS_DIR, f"screenshot_{domain}.json") if os.path.exists(json_path): try: with open(json_path, 'r', encoding='utf-8') as jf: meta = json.load(jf) b64 = meta.get("data_base64") if b64: return (b64, "") # inline base64 present except Exception: pass # Check PNG domain file png_path = os.path.join(OUTPUTS_DIR, f"screenshot_{domain}.png") if os.path.exists(png_path): return ("", png_path) # Fallback to latest latest = os.path.join(OUTPUTS_DIR, "screenshot_latest.png") if os.path.exists(latest): return ("", latest) return ("", "") def build_html_report(title: str, sections: List[dict], summary: dict, screenshot_inline_b64: str = None, screenshot_file: str = None, report_path: str = None) -> str: now = datetime.datetime.utcnow().isoformat() html_parts = [ "", html.escape(title), "" ] html_parts.append(f"

{html.escape(title)}

") html_parts.append(f"
Generated: {now} UTC
") # Summary cards (vertical stack) html_parts.append("
") # Top Issues Card html_parts.append("

Top Issues

") if summary['findings']: for f in summary['findings'][:10]: sev_class = 'high' if f['severity']=='High' else ('medium' if f['severity']=='Medium' else 'low') html_parts.append(f"
{html.escape(f['severity'])} {html.escape(f['script'])}
{html.escape(f['line'][:100])}
") else: html_parts.append("

No notable issues automatically detected.

") html_parts.append("
") # Severity Counts Card html_parts.append("

Counts by Severity

") html_parts.append(f"

● High: {summary['counts'].get('High',0)}

") html_parts.append(f"

● Medium: {summary['counts'].get('Medium',0)}

") html_parts.append(f"

● Low: {summary['counts'].get('Low',0)}

") html_parts.append("
") # Screenshot Card with modal html_parts.append("

Screenshot

") if screenshot_inline_b64: html_parts.append(f"screenshot") html_parts.append("

Click to view full size | Embedded screenshot from outputs/

") elif screenshot_file and os.path.exists(screenshot_file): if report_path: rel_path = os.path.relpath(screenshot_file, start=os.path.dirname(os.path.abspath(report_path))) else: rel_path = os.path.relpath(screenshot_file, start=os.path.dirname(os.path.abspath(__file__))) html_parts.append(f"screenshot") html_parts.append(f"

Click to view full size | Saved: {html.escape(rel_path)}

") else: html_parts.append("

No screenshot available.

") html_parts.append("
") html_parts.append("
") # end summary # LLM Ideas Card (aggregate AI analyses if present) llm_ideas = summary.get('llm_ideas', []) html_parts.append("

LLM Ideas & Analysis

") if llm_ideas: html_parts.append("
    ") for item in llm_ideas: # item expected as dict: {'script':..., 'analysis':...} script = html.escape(item.get('script','')) analysis = html.escape(item.get('analysis','')) html_parts.append(f"
  1. {script}: {analysis}
    2. ") html_parts.append("
") else: html_parts.append("

No LLM analysis available.

") html_parts.append("
") # Full sections with descriptions and suggestions for sec in sections: script_name = sec['name'] description = get_description(script_name) suggestions = sec.get('suggestions', []) html_parts.append(f"

{html.escape(script_name)}

") html_parts.append(f"
{html.escape(description)}
") # Add suggestions box if any suggestions found if suggestions: html_parts.append("
") html_parts.append("

💡 Suggestions & Remediation

") html_parts.append("
    ") for suggestion in suggestions: # Clean up suggestion text (remove numbering if present) clean_suggestion = suggestion.lstrip('0123456789.- ') html_parts.append(f"
  1. {html.escape(clean_suggestion)}
    2. ") html_parts.append("
") html_parts.append("
") # Add AI analysis if present ai_text = sec.get('ai_analysis') if ai_text: html_parts.append("

AI SECURITY ANALYSIS

") # keep it brief in the section view html_parts.append(f"

{html.escape(ai_text[:1000])}

") html_parts.append("
") # Include any saved follow-up Q&A for this script if sec.get('followups'): html_parts.append("

Follow-up Q&A

") for fu in sec.get('followups', []): ts = fu.get('timestamp') try: ts_str = datetime.datetime.utcfromtimestamp(int(ts)).isoformat() + ' UTC' except Exception: ts_str = str(ts) q = html.escape(fu.get('question','')) raw = html.escape((fu.get('response_raw') or '')[:4000]) parsed = fu.get('response_parsed') html_parts.append(f"
{html.escape(ts_str)} — Question: {q}
") if parsed: html_parts.append("
" + html.escape(json.dumps(parsed, indent=2)) + "
") else: html_parts.append("
" + raw + "
") html_parts.append("
") html_parts.append("
") html_parts.append("

Test Output:

") html_parts.append("
")
        html_parts.append(html.escape(sec['output'][:2000]))
        html_parts.append("
") # Modal for full-size screenshot html_parts.append("") # JavaScript for modal functionality html_parts.append("") html_parts.append("") return ''.join(html_parts) def run_screenshot_for_target(target: str) -> None: """ Invoke screenshot_taker.py to ensure a screenshot exists for the target. Prints its stdout/stderr to the console for visibility. """ script = os.path.join(PROJECT_DIR, "screenshot_taker.py") if not os.path.exists(script): print(f"[-] Screenshot script not found: {script}") return try: print(f"[*] Taking screenshot for {target} ...") proc = subprocess.run([sys.executable, script, target], capture_output=True, text=True, timeout=180) if proc.stdout: print(proc.stdout) if proc.stderr: print("--- STDERR from screenshot_taker.py ---") print(proc.stderr) except Exception as e: print(f"[-] Error running screenshot_taker.py: {e}") def main(): parser = argparse.ArgumentParser(description="Run selected scanners and generate a single HTML report.") parser.add_argument("target", help="Target URL or domain (e.g., example.com)") parser.add_argument("-o", "--out", help="Output HTML filename (default: outputs/report_.html)") args = parser.parse_args() target = args.target if '://' in args.target else 'http://' + args.target sections = [] aggregated_findings = [] aggregated_llm_ideas = [] # Ensure outputs dir exists and take a screenshot first os.makedirs(OUTPUTS_DIR, exist_ok=True) run_screenshot_for_target(target) for script in SCRIPTS_TO_RUN: print(f"[*] Running {script} ...") output, findings, suggestions, ai_analysis = run_script_and_capture(script, target) sections.append({'name': script, 'output': output, 'suggestions': suggestions, 'ai_analysis': ai_analysis}) aggregated_findings.extend(findings) if ai_analysis: aggregated_llm_ideas.append({'script': script, 'analysis': ai_analysis}) # Build summary counts = {'High':0,'Medium':0,'Low':0} for f in aggregated_findings: sev = f.get('severity','Low') counts[sev] = counts.get(sev,0) + 1 sorted_findings = sorted(aggregated_findings, key=lambda x: (0 if x['severity']=='High' else (1 if x['severity']=='Medium' else 2), x.get('script',''))) summary = { 'counts': counts, 'findings': sorted_findings } if aggregated_llm_ideas: summary['llm_ideas'] = aggregated_llm_ideas # Load any persisted follow-ups saved during interactive runs (if present) try: parsed_target = urlparse(target) host = (parsed_target.netloc.split(':')[0]) if parsed_target and parsed_target.netloc else re.sub(r'[^A-Za-z0-9_\-]','_', target) followup_file = os.path.join(OUTPUTS_DIR, f'followups_{host}.json') if os.path.exists(followup_file): try: with open(followup_file, 'r', encoding='utf-8') as ff: followups = json.load(ff) # attach followups to corresponding sections for fu in followups: sname = fu.get('script') for sec in sections: if sec['name'] == sname: sec.setdefault('followups', []).append(fu) break # also include in summary for quick access summary['followups'] = followups except Exception: pass except Exception: pass inline_b64, screenshot_file = find_screenshot_for_target(target) safe_name = args.out or os.path.join(OUTPUTS_DIR, f"report_{target.replace('://','_').replace('/','_')}.html") report_html = build_html_report(f"Security Scan Report for {target}", sections, summary, screenshot_inline_b64=inline_b64 if inline_b64 else None, screenshot_file=screenshot_file if screenshot_file else None, report_path=safe_name) try: with open(safe_name, 'w', encoding='utf-8') as f: f.write(report_html) print(f"[+] Report written to {os.path.abspath(safe_name)}") except Exception as e: print(f"[-] Could not write report file: {e}") if __name__ == "__main__": main()