import streamlit as st from neo4j import GraphDatabase from pyvis.network import Network import os import time from dotenv import load_dotenv from pathlib import Path import json import html load_dotenv() BASE_DIR = Path(__file__).parent.parent.resolve() NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687") NEO4J_USER = os.getenv("NEO4J_USER", "neo4j") NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "vulngraph123") from schemas import( parse_bandit_findings, parse_trivy_results, parse_gitleaks_findings ) DEMO_DATA_PATH = BASE_DIR / "app" / "demo_data.json" def load_demo_data() -> dict: """Load static demo data when Neo4j is unavailable.""" try: with open(DEMO_DATA_PATH, "r") as f: return json.load(f) except Exception: return {"findings": [], "stats": {}, "source": "static_demo"} def is_neo4j_available() -> bool: """Check if Neo4j is reachable.""" try: with get_driver().session() as session: session.run("RETURN 1") return True except Exception: return False #Streamlit Page Configuration st.set_page_config( page_title="VulnGraph", page_icon="๐ก๏ธ", layout="wide", initial_sidebar_state="expanded" ) #CSS st.markdown(""" """, unsafe_allow_html=True) #Connect to Neo4j @st.cache_resource def get_driver(): return GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD)) def normalize_path(raw_path: str) -> str: path_str = str(raw_path).replace("\\", "/").strip() base_str = str(BASE_DIR).replace("\\", "/") + "/" if path_str.startswith(base_str): path_str = path_str[len(base_str):] path_str = path_str.lstrip("./") return path_str if path_str else "" # Loading Data def clear_and_load_data(): from scanner import scan_all driver = get_driver() with driver.session() as session: session.run("MATCH (n) DETACH DELETE n") findings = scan_all() with driver.session() as session: for root, _, files in os.walk(BASE_DIR): if any(x in root for x in [".git", "node_modules", ".venv","tools","__pycache__","tmp","data"]): continue for file in files: path=normalize_path(os.path.join(root, file)) if path: session.run("MERGE (f:File {path: $path})", path=path) #Gitleaks git_valid, git_failed= parse_gitleaks_findings(findings["gitleaks"]) for finding in git_valid: file_path=normalize_path(finding.file_path) if not file_path: continue session.run(""" MATCH (f:File {path: $file_path}) MERGE (s:Secret {rule: $rule, source: 'gitleaks'}) ON CREATE SET s.line = $line MERGE (f)-[:CONTAINS]->(s) """, file_path=file_path, rule=finding.finding_id, line=finding.line_number or 0) #BANDIT bandit_valid,bandit_failed=parse_bandit_findings(findings["bandit"]) for finding in bandit_valid: file_path = normalize_path(finding.file_path) if not file_path or ".." in file_path: continue session.run(""" MATCH (f:File {path: $file_path}) MERGE (v:Vulnerability {id: $issue_id, source: 'bandit'}) ON CREATE SET v.severity = $severity, v.text = $text, v.confidence = $confidence,v.cwe=$cwe ON MATCH SET v.severity = $severity MERGE (f)-[:HAS_VULNERABILITY]->(v) """, file_path=file_path, issue_id= finding.finding_id, severity=finding.severity.value, text=finding.text, confidence=finding.confidence or "UNDEFINED", cwe=finding.cwe or "" ) #TRIVY trivy_valid,trivy_failed=parse_trivy_results(findings["trivy"]) for finding in trivy_valid: target_path = normalize_path(finding.file_path) if not target_path or any(x in target_path for x in [".git", ".venv","tools"]): continue session.run("MERGE (f:File {path: $path})", path=target_path) session.run(""" MATCH (f:File {path: $file_path}) MERGE (v:Vulnerability {id: $vuln_id, source: 'trivy'}) ON CREATE SET v.severity = $severity, v.title = $title,v.pkg_name=$pkg_name,v.fixed_version=$fixed_version ON MATCH SET v.severity = $severity MERGE (f)-[:HAS_VULNERABILITY]->(v) """, file_path=target_path, vuln_id=finding.finding_id, severity=finding.severity.value, title=finding.title or "", pkg_name=finding.pkg_name or "", fixed_version=finding.fixed_version or "" ) trivy_secrets = [f for f in trivy_valid if not f.finding_id.startswith("CVE-")] for finding in trivy_secrets: target_path=normalize_path(finding.file_path) if not target_path: continue session.run(""" MATCH (f: File {path: $path}) MERGE (s:Secret {rule: $rule_id, source: 'trivy'}) ON CREATE SET s.match= $match MERGE (f)-[:CONTAINS]->(s) """, path=target_path,rule_id=finding.finding_id,match=finding.text[:100]) return findings #Statistics def get_stats(): if not is_neo4j_available(): demo = load_demo_data() s = demo.get("stats", {}) return { "files": s.get("files", 0), "vulns": s.get("vulns", 0), "secrets": s.get("secrets", 0), "high_critical": s.get("high_critical", 0) } try: with get_driver().session() as session: row = session.run(""" MATCH (f:File) WITH count(f) AS files OPTIONAL MATCH (v:Vulnerability) WITH files, count(v) AS vulns OPTIONAL MATCH (s:Secret) WITH files, vulns, count(s) AS secrets OPTIONAL MATCH (v2:Vulnerability) WHERE toUpper(v2.severity) IN ['CRITICAL', 'HIGH'] RETURN files, vulns, secrets, count(v2) AS high_critical """).single() if row: return dict(row) except Exception: pass return {"files": 0, "vulns": 0, "secrets": 0, "high_critical": 0} def get_severity_breakdown(): try: with get_driver().session() as session: return [(r["severity"], r["count"]) for r in session.run(""" MATCH (v:Vulnerability) RETURN toUpper(v.severity) AS severity, count(v) AS count ORDER BY count DESC """)] except Exception: return [] def get_recent_findings(limit=10): if not is_neo4j_available(): demo = load_demo_data() findings = demo.get("findings", [])[:limit] return [ { "file": f.get("file", ""), "type": f.get("type", ""), "id": f.get("id", ""), "severity": f.get("severity", "UNKNOWN"), "source": f.get("source", "") } for f in findings ] try: with get_driver().session() as session: return [dict(r) for r in session.run(""" MATCH (f:File)-[]->(n) WHERE n:Vulnerability OR n:Secret RETURN f.path AS file, labels(n)[0] AS type, CASE WHEN n:Vulnerability THEN n.id ELSE n.rule END AS id, CASE WHEN n:Vulnerability THEN n.severity ELSE 'SECRET' END AS severity, n.source AS source LIMIT $limit """, limit=limit)] except Exception: return [] @st.cache_data(ttl=10) def ollama_status(): from llm import check_ollama_health return check_ollama_health() def get_explained_findings(limit = 5): """Fetch nodes which have LLM explanations.""" if not is_neo4j_available(): demo = load_demo_data() explained = [f for f in demo.get("findings", []) if f.get("explanation")][:limit] return explained try: with get_driver().session() as session: return [dict(r) for r in session.run(""" MATCH (n) WHERE (n:Vulnerability OR n:Secret) AND n.explanation IS NOT NULL RETURN labels(n)[0] AS type, CASE WHEN n:Vulnerability THEN n.id ELSE n.rule END AS id, CASE WHEN n:Vulnerability THEN n.severity ELSE 'SECRET' END AS severity, n.explanation AS explanation, n.why_dangerous AS why_dangerous, n.fix AS fix, n.cwe AS cwe LIMIT $limit """, limit=limit)] except Exception: return [] #Creating the Graph def generate_graph(): try: with get_driver().session() as session: result = list(session.run("MATCH (n)-[r]->(m) RETURN n, r, m")) isolated =list(session.run("MATCH (n) WHERE NOT (n)--() RETURN n")) except Exception: return None if not result and not isolated: return None net = Network(height="650px", width="100%", directed=True, bgcolor="#0d1117", font_color="#c9d1d9") net.toggle_physics(True) net.set_options(""" { "edges": { "arrows": {"to": {"enabled": true, "scaleFactor": 1.5}}, "color": {"color": "#1e3a5f", "highlight": "#4a90d9"}, "width": 1.8, "smooth": {"enabled": true, "type": "continuous"} }, "nodes": { "font": {"color": "#c9d1d9", "size": 13, "face": "monospace"}, "borderWidth": 2 }, "physics": { "enabled": true, "barnesHut": {"gravitationalConstant": -7000, "springLength": 200, "damping": 0.15} }, "interaction": {"hover": true, "tooltipDelay": 100, "hideEdgesOnDrag": true} } """) SEV_COLORS = {"CRITICAL": "#ff4444", "HIGH": "#ff8c00", "MEDIUM": "#ffd700", "LOW": "#4fc3a1"} nodes_added = set() def add_node(node): nid = node.element_id if nid in nodes_added: return label = list(node.labels)[0] source= node.get("source", "") if label == "File": path= node.get("path", "") short = path.split("/")[-1] or path title = f" {path}" display = f"{short}" color = {"background": "#1a2744", "border": "#2d5a8e", "highlight": {"border": "#00c9ff"}} size = 18 elif label == "Secret": rule = node.get("rule", "?") title = f"Secret | Rule: {rule} | Line: {node.get('line','?')} | Source: {source}" display = f"{rule[:20]}" color = {"background": "#3d1a1a", "border": "#e06c75", "highlight": {"border": "#ff4444"}} size = 22 else: # Vulnerability vid = node.get("id", "?") sev = (node.get("severity") or "UNKNOWN").upper() sev_color = SEV_COLORS.get(sev, "#8b949e") text = node.get("text") or node.get("title") or "" explanation = node.get("explanation", "") title = f"{vid} | {sev} | {source}" if text: title += f"\n{text[:120]}" if explanation: title += f"\n\n{explanation[:200]}" display = f"{vid[:18]}" color = {"background": "#2a1f0a", "border": sev_color, "highlight": {"border": sev_color}} size = 26 if sev in ["CRITICAL", "HIGH"] else 20 net.add_node(nid, label=display, title=title, color=color, size=size) nodes_added.add(nid) for record in result: add_node(record["n"]) add_node(record["m"]) r = record["r"] net.add_edge(r.start_node.element_id, r.end_node.element_id, color="#1e3a5f", width=2, arrows="to") for record in isolated: add_node(record["n"]) os.makedirs("tmp", exist_ok=True) path="tmp/vulngraph.html" net.save_graph(path) with open(path,"r", encoding="utf-8") as f: html = f.read() html = html.replace("
", "") html = html.replace("background-color:#ffffff", "background-color:#0d1117") with open(path, "w", encoding="utf-8") as f: f.write(html) with open(path, "r", encoding="utf-8") as f: return f.read() #Sidebar def render_sidebar(stats, severity_breakdown): with st.sidebar: st.markdown("""