import os import sys import json import uuid from pathlib import Path PROJECT_ROOT = Path(__file__).parent.parent.parent sys.path.insert(0, str(PROJECT_ROOT)) from recon.partial_recon_modules.helpers import ( _classify_ip, _is_ip_or_cidr, _is_valid_hostname, _is_valid_url, _resolve_hostname, _should_include_root_domain, ) from recon.partial_recon_modules.graph_builders import _build_vuln_scan_data_from_graph def run_nuclei(config: dict) -> None: """ Run partial vulnerability scanning using Nuclei. Scans BaseURLs and endpoints from the graph (and/or user-provided URLs) with the Nuclei vulnerability scanner. Results are merged into the graph via update_graph_from_vuln_scan. """ from recon.main_recon_modules.vuln_scan import run_vuln_scan from recon.project_settings import get_settings domain = config["domain"] user_id = os.environ.get("USER_ID", "") project_id = os.environ.get("PROJECT_ID", "") print(f"[*][Partial Recon] Loading project settings...") settings = get_settings() # Force-enable Nuclei since the user explicitly chose to run it settings['NUCLEI_ENABLED'] = True # Apply settings_overrides from modal checkboxes (bypass DB settings) settings_overrides = config.get("settings_overrides") or {} for key, value in settings_overrides.items(): settings[key] = value print(f"\n{'=' * 50}") print(f"[*][Partial Recon] Nuclei Vulnerability Scanning") print(f"[*][Partial Recon] Domain: {domain}") print(f"{'=' * 50}\n") # Parse user targets -- Nuclei accepts URLs user_targets = config.get("user_targets") or {} user_urls = [] url_attach_to = None if user_targets: for entry in user_targets.get("urls", []): entry = entry.strip() if entry and _is_valid_url(entry): user_urls.append(entry) elif entry: print(f"[!][Partial Recon] Skipping invalid URL: {entry}") url_attach_to = user_targets.get("url_attach_to") # BaseURL or None if user_urls: print(f"[+][Partial Recon] Validated {len(user_urls)} custom URLs") if url_attach_to: print(f"[+][Partial Recon] URLs will be attached to BaseURL: {url_attach_to}") else: print(f"[+][Partial Recon] URLs will be tracked via UserInput (generic)") # Track whether we need a UserInput node needs_user_input = bool(user_urls and not url_attach_to) # Build recon_data from Neo4j graph (or start empty if user unchecked graph targets) include_graph = config.get("include_graph_targets", True) if include_graph: print(f"[*][Partial Recon] Querying graph for targets (BaseURLs, Endpoints, DNS)...") recon_data = _build_vuln_scan_data_from_graph( domain, user_id, project_id, include_root_domain=_should_include_root_domain(settings), ) else: print(f"[*][Partial Recon] Skipping graph targets (user opted out)") recon_data = { "domain": domain, "subdomains": [], "dns": { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }, "http_probe": { "by_url": {}, }, "resource_enum": { "by_base_url": {}, "discovered_urls": [], }, } # Inject user-provided URLs into http_probe targets if user_urls: print(f"[*][Partial Recon] Adding {len(user_urls)} user-provided URLs to scan targets") for url in user_urls: if url not in recon_data["http_probe"]["by_url"]: from urllib.parse import urlparse parsed = urlparse(url) recon_data["http_probe"]["by_url"][url] = { "url": url, "host": parsed.netloc.split(":")[0], "status_code": 200, "content_type": "text/html", } # Also add user URLs with parameters to DAST list if "?" in url and "=" in url: if url not in recon_data["resource_enum"]["discovered_urls"]: recon_data["resource_enum"]["discovered_urls"].append(url) # Ensure all target hostnames are in subdomains list for scope filtering existing_subs = set(recon_data.get("subdomains", [])) for url_key, url_data in recon_data["http_probe"]["by_url"].items(): host = url_data.get("host", "") if host and host not in existing_subs: existing_subs.add(host) recon_data["subdomains"] = list(existing_subs) # Check we have something to scan has_http_targets = bool(recon_data["http_probe"]["by_url"]) has_dns_targets = bool(recon_data["dns"]["domain"]["has_records"] or recon_data["dns"]["subdomains"]) if not has_http_targets and not has_dns_targets: print("[!][Partial Recon] No targets to scan (no BaseURLs, endpoints, or DNS data in graph).") print("[!][Partial Recon] Run HTTP Probing (Httpx) or Subdomain Discovery first, or provide URLs manually.") sys.exit(1) # Run vuln_scan (same function as the full pipeline) print(f"[*][Partial Recon] Running Nuclei vulnerability scan...") result = run_vuln_scan(recon_data, settings=settings) # Also run MITRE enrichment if enabled if settings.get('MITRE_ENABLED', True): try: from recon.main_recon_modules.add_mitre import run_mitre_enrichment print(f"[*][Partial Recon] Running MITRE ATT&CK enrichment...") result = run_mitre_enrichment(result, settings=settings) except Exception as e: print(f"[!][Partial Recon] MITRE enrichment failed: {e}") # Pre-create BaseURL + UserInput BEFORE graph update so vuln_scan can find and attach to them if user_urls: print(f"[*][Partial Recon] Pre-creating BaseURL nodes for user-provided URLs...") try: from graph_db import Neo4jClient from urllib.parse import urlparse as _urlparse with Neo4jClient() as graph_client: if graph_client.verify_connection(): driver = graph_client.driver with driver.session() as session: if url_attach_to: # Verify the attachment BaseURL still exists check = session.run( """ MATCH (b:BaseURL {url: $url, user_id: $uid, project_id: $pid}) RETURN b.url AS url """, url=url_attach_to, uid=user_id, pid=project_id, ) if check.single(): print(f"[+][Partial Recon] BaseURL {url_attach_to} exists, linking user URLs") else: print(f"[!][Partial Recon] BaseURL {url_attach_to} not found, falling back to UserInput") needs_user_input = True url_attach_to = None if needs_user_input: import uuid user_input_id = str(uuid.uuid4()) session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) ON CREATE SET ui.type = 'url', ui.values = $urls, ui.created_at = datetime(), ui.tool = 'Nuclei' SET ui.updated_at = datetime() MERGE (d)-[:HAS_USER_INPUT]->(ui) """, domain=domain, uid=user_id, pid=project_id, ui_id=user_input_id, urls=user_urls, ) # Create BaseURL nodes for each user URL (so vuln_scan attaches findings here) for url in user_urls: parsed = _urlparse(url) base_url = f"{parsed.scheme}://{parsed.netloc}" if needs_user_input: session.run( """ MATCH (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) MERGE (b:BaseURL {url: $url, user_id: $uid, project_id: $pid}) ON CREATE SET b.host = $host, b.source = 'partial_recon_user_input', b.updated_at = datetime() MERGE (ui)-[:PRODUCED]->(b) """, ui_id=user_input_id, uid=user_id, pid=project_id, url=base_url, host=parsed.netloc.split(":")[0], ) else: session.run( """ MERGE (b:BaseURL {url: $url, user_id: $uid, project_id: $pid}) ON CREATE SET b.host = $host, b.source = 'partial_recon_user_input', b.updated_at = datetime() """, uid=user_id, pid=project_id, url=base_url, host=parsed.netloc.split(":")[0], ) print(f"[+][Partial Recon] Pre-created BaseURL + UserInput nodes for user URLs") except Exception as e: print(f"[!][Partial Recon] Failed to pre-create BaseURL nodes: {e}") # Update the graph database (vuln_scan will find the BaseURLs and attach findings to them) print(f"[*][Partial Recon] Updating graph database...") try: from graph_db import Neo4jClient with Neo4jClient() as graph_client: if graph_client.verify_connection(): stats = graph_client.update_graph_from_vuln_scan( recon_data=result, user_id=user_id, project_id=project_id, ) print(f"[+][Partial Recon] Graph updated successfully") print(f"[+][Partial Recon] Stats: {json.dumps(stats, default=str)}") else: print("[!][Partial Recon] Neo4j not reachable, graph not updated") except Exception as e: print(f"[!][Partial Recon] Graph update failed: {e}") raise print(f"\n[+][Partial Recon] Nuclei vulnerability scanning completed successfully") def run_subdomain_takeover_partial(config: dict) -> None: """ Partial subdomain takeover scan. Accepts user-provided custom subdomains (validated against the project domain) in `user_targets.subdomains`. Reuses the full pipeline's runner (run_subdomain_takeover) and writes Vulnerability nodes with source="takeover_scan" via update_graph_from_subdomain_takeover. """ from recon.main_recon_modules.subdomain_takeover import run_subdomain_takeover from recon.project_settings import get_settings domain = config["domain"] user_id = os.environ.get("USER_ID", "") project_id = os.environ.get("PROJECT_ID", "") print(f"[*][Partial Recon] Loading project settings...") settings = get_settings() # Force-enable takeover since the user explicitly chose to run it settings['SUBDOMAIN_TAKEOVER_ENABLED'] = True settings_overrides = config.get("settings_overrides") or {} for key, value in settings_overrides.items(): settings[key] = value print(f"\n{'=' * 50}") print(f"[*][Partial Recon] Subdomain Takeover") print(f"[*][Partial Recon] Domain: {domain}") print(f"{'=' * 50}\n") # --- User-provided subdomains --- user_targets = config.get("user_targets") or {} user_subdomains: list[str] = [] resolved_sub_ips: dict = {} for entry in (user_targets.get("subdomains") or []): entry = (entry or "").strip().lower() if not entry: continue if not _is_valid_hostname(entry): print(f"[!][Partial Recon] Skipping invalid hostname: {entry}") continue if not (entry == domain or entry.endswith("." + domain)): print(f"[!][Partial Recon] Skipping out-of-scope subdomain: {entry}") continue user_subdomains.append(entry) if user_subdomains: print(f"[+][Partial Recon] Validated {len(user_subdomains)} custom subdomain(s)") # --- Build recon_data from graph (or empty) --- include_graph = config.get("include_graph_targets", True) if include_graph: print(f"[*][Partial Recon] Querying graph for targets (Subdomains, DNS, alive URLs)...") recon_data = _build_vuln_scan_data_from_graph( domain, user_id, project_id, include_root_domain=_should_include_root_domain(settings), ) else: print(f"[*][Partial Recon] Skipping graph targets (user opted out)") recon_data = { "domain": domain, "subdomains": [], "dns": { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }, "http_probe": {"by_url": {}, "live_urls": []}, } # Ensure keys the runner reads recon_data.setdefault("domain", domain) recon_data.setdefault("subdomains", []) recon_data.setdefault("dns", { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }) recon_data.setdefault("http_probe", {"by_url": {}, "live_urls": []}) # --- Resolve & inject user subdomains --- if user_subdomains: print(f"[*][Partial Recon] Resolving {len(user_subdomains)} custom subdomain(s)...") for sub in user_subdomains: if sub not in recon_data["subdomains"]: recon_data["subdomains"].append(sub) if sub not in recon_data["dns"]["subdomains"]: ips = _resolve_hostname(sub) all_ips = ips.get("ipv4", []) + ips.get("ipv6", []) recon_data["dns"]["subdomains"][sub] = { "ips": ips, "has_records": bool(all_ips), } if all_ips: resolved_sub_ips[sub] = ips print(f"[+][Partial Recon] Resolved {sub} -> {all_ips}") else: # Still worth scanning — a dangling CNAME is exactly what we're looking for print(f"[*][Partial Recon] {sub} has no A/AAAA records (potential takeover candidate)") # Need at least one subdomain to scan if not recon_data["subdomains"] and not recon_data["dns"]["subdomains"]: print("[!][Partial Recon] No subdomains to scan (graph empty and no custom targets).") sys.exit(1) # --- Run the scanner --- print(f"[*][Partial Recon] Running subdomain takeover scan...") run_subdomain_takeover(recon_data, settings=settings) findings = recon_data.get("subdomain_takeover", {}).get("findings", []) summary = recon_data.get("subdomain_takeover", {}).get("summary", {}) print(f"[+][Partial Recon] Takeover scan complete: {summary}") # --- Write findings + link user-provided subdomains to the graph --- print(f"[*][Partial Recon] Updating graph database...") try: from graph_db import Neo4jClient with Neo4jClient() as graph_client: if not graph_client.verify_connection(): print("[!][Partial Recon] Neo4j not reachable, graph not updated") return # Ensure user subdomains exist in graph so HAS_VULNERABILITY can attach if user_subdomains: driver = graph_client.driver with driver.session() as session: for sub in user_subdomains: session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) ON CREATE SET s.source = 'partial_recon_user_input', s.created_at = datetime() SET s.updated_at = datetime() MERGE (d)-[:HAS_SUBDOMAIN]->(s) """, domain=domain, uid=user_id, pid=project_id, sub=sub, ) for bucket, addrs in (resolved_sub_ips.get(sub) or {}).items(): for addr in addrs or []: session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) MERGE (s)-[:RESOLVES_TO]->(i) """, addr=addr, uid=user_id, pid=project_id, version=bucket, sub=sub, ) stats = graph_client.update_graph_from_subdomain_takeover( recon_data=recon_data, user_id=user_id, project_id=project_id, ) print(f"[+][Partial Recon] Graph updated: {json.dumps(stats, default=str)}") except Exception as e: print(f"[!][Partial Recon] Graph update failed: {e}") raise print(f"\n[+][Partial Recon] Subdomain takeover scan completed successfully ({len(findings)} finding(s))") def run_security_checks_partial(config: dict) -> None: """ Run partial security checks (Direct IP Access, TLS/SSL, Security Headers, DNS, etc.) Uses the same run_security_checks() function from the full pipeline. Targets are loaded from the graph (IPs, subdomains, BaseURLs) and/or user-provided custom subdomains, IPs, and URLs. Results are stored as Vulnerability nodes via update_graph_from_vuln_scan. """ from recon.helpers import run_security_checks from recon.project_settings import get_settings domain = config["domain"] user_id = os.environ.get("USER_ID", "") project_id = os.environ.get("PROJECT_ID", "") print(f"[*][Partial Recon] Loading project settings...") settings = get_settings() # Force-enable security checks since the user explicitly chose to run them settings['SECURITY_CHECK_ENABLED'] = True print(f"\n{'=' * 50}") print(f"[*][Partial Recon] Security Checks") print(f"[*][Partial Recon] Domain: {domain}") print(f"{'=' * 50}\n") # Parse user targets user_targets = config.get("user_targets") or {} user_subdomains = [] user_ips = [] user_urls = [] ip_attach_to = None url_attach_to = None if user_targets: # Validate subdomains for entry in user_targets.get("subdomains", []): entry = entry.strip().lower() if entry and _is_valid_hostname(entry): if entry.endswith("." + domain) or entry == domain: user_subdomains.append(entry) else: print(f"[!][Partial Recon] Skipping out-of-scope subdomain: {entry}") elif entry: print(f"[!][Partial Recon] Skipping invalid hostname: {entry}") # Validate IPs for entry in user_targets.get("ips", []): entry = entry.strip() if entry and _is_ip_or_cidr(entry): user_ips.append(entry) elif entry: print(f"[!][Partial Recon] Skipping invalid IP: {entry}") ip_attach_to = user_targets.get("ip_attach_to") # Validate URLs for entry in user_targets.get("urls", []): entry = entry.strip() if entry and _is_valid_url(entry): user_urls.append(entry) elif entry: print(f"[!][Partial Recon] Skipping invalid URL: {entry}") url_attach_to = user_targets.get("url_attach_to") if user_subdomains: print(f"[+][Partial Recon] Validated {len(user_subdomains)} custom subdomains") if user_ips: print(f"[+][Partial Recon] Validated {len(user_ips)} custom IPs") if ip_attach_to: print(f"[+][Partial Recon] IPs will be attached to subdomain: {ip_attach_to}") if user_urls: print(f"[+][Partial Recon] Validated {len(user_urls)} custom URLs") if url_attach_to: print(f"[+][Partial Recon] URLs will be attached to BaseURL: {url_attach_to}") # Track whether we need UserInput nodes needs_ip_user_input = bool(user_ips and not ip_attach_to) needs_url_user_input = bool(user_urls and not url_attach_to) # Build recon_data from Neo4j graph include_graph = config.get("include_graph_targets", True) if include_graph: print(f"[*][Partial Recon] Querying graph for targets (IPs, Subdomains, BaseURLs, DNS)...") recon_data = _build_vuln_scan_data_from_graph( domain, user_id, project_id, include_root_domain=_should_include_root_domain(settings), ) else: print(f"[*][Partial Recon] Skipping graph targets (user opted out)") recon_data = { "domain": domain, "subdomains": [], "dns": { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }, "http_probe": { "by_url": {}, }, "resource_enum": { "by_base_url": {}, "discovered_urls": [], }, } # STEP 1: Resolve user subdomains and inject into recon_data DNS resolved_sub_ips = {} if user_subdomains: print(f"[*][Partial Recon] Resolving {len(user_subdomains)} custom subdomains...") for sub in user_subdomains: ips = _resolve_hostname(sub) all_ips = ips.get("ipv4", []) + ips.get("ipv6", []) if all_ips: resolved_sub_ips[sub] = ips if sub not in recon_data["dns"]["subdomains"]: recon_data["dns"]["subdomains"][sub] = { "ips": ips, "has_records": True, } else: for bucket in ("ipv4", "ipv6"): for addr in ips.get(bucket, []): if addr not in recon_data["dns"]["subdomains"][sub]["ips"][bucket]: recon_data["dns"]["subdomains"][sub]["ips"][bucket].append(addr) if sub not in recon_data.get("subdomains", []): recon_data.setdefault("subdomains", []).append(sub) print(f"[+][Partial Recon] Resolved {sub} -> {all_ips}") else: print(f"[!][Partial Recon] Could not resolve {sub}") # STEP 2: Inject user IPs into recon_data DNS if user_ips: print(f"[*][Partial Recon] Adding {len(user_ips)} user-provided IPs to DNS data") for ip_str in user_ips: bucket = _classify_ip(ip_str) if ip_attach_to: # Inject into subdomain's IP list if ip_attach_to not in recon_data["dns"]["subdomains"]: recon_data["dns"]["subdomains"][ip_attach_to] = { "ips": {"ipv4": [], "ipv6": []}, "has_records": True, } if ip_str not in recon_data["dns"]["subdomains"][ip_attach_to]["ips"][bucket]: recon_data["dns"]["subdomains"][ip_attach_to]["ips"][bucket].append(ip_str) else: # Inject into domain's IP list if ip_str not in recon_data["dns"]["domain"]["ips"][bucket]: recon_data["dns"]["domain"]["ips"][bucket].append(ip_str) recon_data["dns"]["domain"]["has_records"] = True # STEP 3: Inject user URLs into http_probe targets if user_urls: print(f"[*][Partial Recon] Adding {len(user_urls)} user-provided URLs to scan targets") for url in user_urls: if url not in recon_data["http_probe"]["by_url"]: from urllib.parse import urlparse parsed = urlparse(url) recon_data["http_probe"]["by_url"][url] = { "url": url, "host": parsed.netloc.split(":")[0], "status_code": 200, "content_type": "text/html", } # Check we have something to scan has_http_targets = bool(recon_data["http_probe"]["by_url"]) has_dns_targets = bool(recon_data["dns"]["domain"]["has_records"] or recon_data["dns"]["subdomains"]) if not has_http_targets and not has_dns_targets: print("[!][Partial Recon] No targets to scan (no IPs, subdomains, or BaseURLs in graph).") print("[!][Partial Recon] Run Subdomain Discovery and HTTP Probing first, or provide custom targets.") sys.exit(1) # Build enabled checks dict from settings security_checks_enabled = { "direct_ip_http": settings.get('SECURITY_CHECK_DIRECT_IP_HTTP', True), "direct_ip_https": settings.get('SECURITY_CHECK_DIRECT_IP_HTTPS', True), "ip_api_exposed": settings.get('SECURITY_CHECK_IP_API_EXPOSED', True), "waf_bypass": settings.get('SECURITY_CHECK_WAF_BYPASS', True), "tls_expiring_soon": settings.get('SECURITY_CHECK_TLS_EXPIRING_SOON', True), "missing_referrer_policy": settings.get('SECURITY_CHECK_MISSING_REFERRER_POLICY', True), "missing_permissions_policy": settings.get('SECURITY_CHECK_MISSING_PERMISSIONS_POLICY', True), "missing_coop": settings.get('SECURITY_CHECK_MISSING_COOP', True), "missing_corp": settings.get('SECURITY_CHECK_MISSING_CORP', True), "missing_coep": settings.get('SECURITY_CHECK_MISSING_COEP', True), "cache_control_missing": settings.get('SECURITY_CHECK_CACHE_CONTROL_MISSING', True), "login_no_https": settings.get('SECURITY_CHECK_LOGIN_NO_HTTPS', True), "session_no_secure": settings.get('SECURITY_CHECK_SESSION_NO_SECURE', True), "session_no_httponly": settings.get('SECURITY_CHECK_SESSION_NO_HTTPONLY', True), "basic_auth_no_tls": settings.get('SECURITY_CHECK_BASIC_AUTH_NO_TLS', True), "spf_missing": settings.get('SECURITY_CHECK_SPF_MISSING', True), "dmarc_missing": settings.get('SECURITY_CHECK_DMARC_MISSING', True), "dnssec_missing": settings.get('SECURITY_CHECK_DNSSEC_MISSING', True), "zone_transfer": settings.get('SECURITY_CHECK_ZONE_TRANSFER', True), "admin_port_exposed": settings.get('SECURITY_CHECK_ADMIN_PORT_EXPOSED', True), "database_exposed": settings.get('SECURITY_CHECK_DATABASE_EXPOSED', True), "redis_no_auth": settings.get('SECURITY_CHECK_REDIS_NO_AUTH', True), "kubernetes_api_exposed": settings.get('SECURITY_CHECK_KUBERNETES_API_EXPOSED', True), "smtp_open_relay": settings.get('SECURITY_CHECK_SMTP_OPEN_RELAY', True), "csp_unsafe_inline": settings.get('SECURITY_CHECK_CSP_UNSAFE_INLINE', True), "insecure_form_action": settings.get('SECURITY_CHECK_INSECURE_FORM_ACTION', True), "no_rate_limiting": settings.get('SECURITY_CHECK_NO_RATE_LIMITING', True), } enabled_count = sum(1 for v in security_checks_enabled.values() if v) print(f"[*][Partial Recon] {enabled_count}/{len(security_checks_enabled)} security checks enabled") if not any(security_checks_enabled.values()): print("[!][Partial Recon] All individual security checks are disabled in settings.") print("[!][Partial Recon] Enable at least one check category to run security checks.") sys.exit(1) # Run security checks (same function as the full pipeline) print(f"[*][Partial Recon] Running security checks...") security_results = run_security_checks( recon_data=recon_data, enabled_checks=security_checks_enabled, timeout=settings.get('SECURITY_CHECK_TIMEOUT', 10), tls_expiry_days=settings.get('SECURITY_CHECK_TLS_EXPIRY_DAYS', 30), max_workers=settings.get('SECURITY_CHECK_MAX_WORKERS', 10), ai_classifier_enabled=settings.get('WAF_AI_CLASSIFIER', False), ai_model=settings.get('AI_PIPELINE_MODEL', 'claude-opus-4-6'), ai_user_id=os.environ.get('USER_ID', ''), ai_project_id=os.environ.get('PROJECT_ID', ''), ) # Merge security checks into recon_data for graph update if "vuln_scan" in recon_data: recon_data["vuln_scan"]["security_checks"] = security_results.get("security_checks", {}) else: recon_data["vuln_scan"] = {"security_checks": security_results.get("security_checks", {})} findings = security_results.get("security_checks", {}).get("findings", []) print(f"[+][Partial Recon] Security checks completed: {len(findings)} findings") # Update the graph database print(f"[*][Partial Recon] Updating graph database...") try: from graph_db import Neo4jClient with Neo4jClient() as graph_client: if graph_client.verify_connection(): stats = graph_client.update_graph_from_vuln_scan( recon_data=recon_data, user_id=user_id, project_id=project_id, ) # Link user-provided subdomains to graph if user_subdomains and resolved_sub_ips: driver = graph_client.driver with driver.session() as session: for sub, ips in resolved_sub_ips.items(): session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) ON CREATE SET s.source = 'partial_recon_user_input', s.created_at = datetime() SET s.updated_at = datetime() MERGE (d)-[:HAS_SUBDOMAIN]->(s) """, domain=domain, uid=user_id, pid=project_id, sub=sub, ) for bucket in ("ipv4", "ipv6"): for addr in ips.get(bucket, []): session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) MERGE (s)-[:RESOLVES_TO]->(i) """, addr=addr, uid=user_id, pid=project_id, version=bucket, sub=sub, ) print(f"[+][Partial Recon] Linked {len(resolved_sub_ips)} user subdomains to graph") # Link user-provided IPs to graph if user_ips: driver = graph_client.driver with driver.session() as session: if ip_attach_to: # Check if subdomain exists check = session.run( """ MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) RETURN s.name AS name """, sub=ip_attach_to, uid=user_id, pid=project_id, ) if check.single(): for ip_str in user_ips: version = _classify_ip(ip_str) session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) MERGE (s)-[:RESOLVES_TO]->(i) """, addr=ip_str, uid=user_id, pid=project_id, version=version, sub=ip_attach_to, ) print(f"[+][Partial Recon] Linked {len(user_ips)} IPs to {ip_attach_to}") else: print(f"[!][Partial Recon] Subdomain {ip_attach_to} not found, falling back to UserInput") needs_ip_user_input = True if needs_ip_user_input: user_input_id = str(uuid.uuid4()) session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) ON CREATE SET ui.type = 'ip', ui.values = $ips, ui.created_at = datetime(), ui.tool = 'SecurityChecks' SET ui.updated_at = datetime() MERGE (d)-[:HAS_USER_INPUT]->(ui) """, domain=domain, uid=user_id, pid=project_id, ui_id=user_input_id, ips=user_ips, ) for ip_str in user_ips: version = _classify_ip(ip_str) session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) MERGE (ui)-[:PRODUCED]->(i) """, addr=ip_str, uid=user_id, pid=project_id, version=version, ui_id=user_input_id, ) print(f"[+][Partial Recon] Created UserInput + linked {len(user_ips)} IPs via PRODUCED") # Link user-provided URLs to graph if user_urls: driver = graph_client.driver with driver.session() as session: if url_attach_to: check = session.run( """ MATCH (b:BaseURL {url: $url, user_id: $uid, project_id: $pid}) RETURN b.url AS url """, url=url_attach_to, uid=user_id, pid=project_id, ) if check.single(): print(f"[+][Partial Recon] BaseURL {url_attach_to} exists, linking user URLs") else: print(f"[!][Partial Recon] BaseURL {url_attach_to} not found, falling back to UserInput") needs_url_user_input = True url_attach_to = None if needs_url_user_input: user_input_id = str(uuid.uuid4()) session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) ON CREATE SET ui.type = 'url', ui.values = $urls, ui.created_at = datetime(), ui.tool = 'SecurityChecks' SET ui.updated_at = datetime() MERGE (d)-[:HAS_USER_INPUT]->(ui) """, domain=domain, uid=user_id, pid=project_id, ui_id=user_input_id, urls=user_urls, ) for url in user_urls: session.run( """ MATCH (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) MATCH (b:BaseURL {url: $url, user_id: $uid, project_id: $pid}) MERGE (ui)-[:PRODUCED]->(b) """, ui_id=user_input_id, uid=user_id, pid=project_id, url=url, ) print(f"[+][Partial Recon] Created UserInput + linked user URLs via PRODUCED") print(f"[+][Partial Recon] Graph updated successfully") print(f"[+][Partial Recon] Stats: {json.dumps(stats, default=str)}") else: print("[!][Partial Recon] Neo4j not reachable, graph not updated") except Exception as e: print(f"[!][Partial Recon] Graph update failed: {e}") raise print(f"\n[+][Partial Recon] Security checks completed successfully") def run_vhost_sni_partial(config: dict) -> None: """ Partial VHost & SNI enumeration. Accepts user-provided custom subdomains (added as candidate hostnames) and custom IPs (added as extra target IPs to probe). Reuses the full pipeline's runner (run_vhost_sni_enrichment) and writes Vulnerability nodes with source="vhost_sni_enum" via update_graph_from_vhost_sni. """ from recon.main_recon_modules.vhost_sni_enum import run_vhost_sni_enrichment from recon.project_settings import get_settings domain = config["domain"] user_id = os.environ.get("USER_ID", "") project_id = os.environ.get("PROJECT_ID", "") print(f"[*][Partial Recon] Loading project settings...") settings = get_settings() # Force-enable since the user explicitly chose to run it settings['VHOST_SNI_ENABLED'] = True settings_overrides = config.get("settings_overrides") or {} for key, value in settings_overrides.items(): settings[key] = value print(f"\n{'=' * 50}") print(f"[*][Partial Recon] VHost & SNI Enumeration") print(f"[*][Partial Recon] Domain: {domain}") print(f"{'=' * 50}\n") # --- User-provided inputs --- user_targets = config.get("user_targets") or {} user_subdomains: list[str] = [] user_ips: list[str] = [] ip_attach_to = user_targets.get("ip_attach_to") resolved_sub_ips: dict = {} for entry in (user_targets.get("subdomains") or []): entry = (entry or "").strip().lower() if not entry: continue if not _is_valid_hostname(entry): print(f"[!][Partial Recon] Skipping invalid hostname: {entry}") continue if not (entry == domain or entry.endswith("." + domain)): print(f"[!][Partial Recon] Skipping out-of-scope subdomain: {entry}") continue user_subdomains.append(entry) for entry in (user_targets.get("ips") or []): entry = (entry or "").strip() if entry and _is_ip_or_cidr(entry): user_ips.append(entry) elif entry: print(f"[!][Partial Recon] Skipping invalid IP: {entry}") if user_subdomains: print(f"[+][Partial Recon] Validated {len(user_subdomains)} custom subdomain(s) as candidate hostnames") if user_ips: print(f"[+][Partial Recon] Validated {len(user_ips)} custom IP(s) as targets") # --- Build recon_data from graph --- include_graph = config.get("include_graph_targets", True) if include_graph: print(f"[*][Partial Recon] Querying graph for targets (Subdomains, IPs, Ports, BaseURLs, DNS)...") recon_data = _build_vuln_scan_data_from_graph( domain, user_id, project_id, include_root_domain=_should_include_root_domain(settings), ) else: print(f"[*][Partial Recon] Skipping graph targets (user opted out)") recon_data = { "domain": domain, "subdomains": [], "dns": { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }, "http_probe": {"by_url": {}, "by_host": {}, "live_urls": []}, "port_scan": {"by_host": {}}, } recon_data.setdefault("domain", domain) recon_data.setdefault("subdomains", []) recon_data.setdefault("dns", { "domain": {"ips": {"ipv4": [], "ipv6": []}, "has_records": False}, "subdomains": {}, }) recon_data.setdefault("http_probe", {"by_url": {}, "by_host": {}, "live_urls": []}) recon_data.setdefault("port_scan", {"by_host": {}}) # --- Resolve user subdomains and inject --- if user_subdomains: print(f"[*][Partial Recon] Resolving {len(user_subdomains)} custom subdomain(s)...") for sub in user_subdomains: if sub not in recon_data["subdomains"]: recon_data["subdomains"].append(sub) if sub not in recon_data["dns"]["subdomains"]: ips = _resolve_hostname(sub) all_ips = ips.get("ipv4", []) + ips.get("ipv6", []) recon_data["dns"]["subdomains"][sub] = { "ips": ips, "has_records": bool(all_ips), } if all_ips: resolved_sub_ips[sub] = ips print(f"[+][Partial Recon] Resolved {sub} -> {all_ips}") # Add resolved IPs as targets too (so wordlist probes hit them) for ip in all_ips: recon_data["port_scan"]["by_host"].setdefault(ip, { "ip": ip, "ports": [{"port": 443, "protocol": "tcp"}, {"port": 80, "protocol": "tcp"}], }) # --- Inject user IPs as targets --- if user_ips: print(f"[*][Partial Recon] Adding {len(user_ips)} user-provided IPs as scan targets") for ip_str in user_ips: recon_data["port_scan"]["by_host"].setdefault(ip_str, { "ip": ip_str, "ports": [{"port": 443, "protocol": "tcp"}, {"port": 80, "protocol": "tcp"}], }) # If attached to a subdomain, also wire DNS so the candidate set picks it up if ip_attach_to: bucket = _classify_ip(ip_str) if ip_attach_to not in recon_data["dns"]["subdomains"]: recon_data["dns"]["subdomains"][ip_attach_to] = { "ips": {"ipv4": [], "ipv6": []}, "has_records": True, } if ip_str not in recon_data["dns"]["subdomains"][ip_attach_to]["ips"][bucket]: recon_data["dns"]["subdomains"][ip_attach_to]["ips"][bucket].append(ip_str) # Need at least one IP target if not recon_data["port_scan"]["by_host"] and not recon_data["dns"]["subdomains"]: print("[!][Partial Recon] No IP targets to probe (graph empty and no custom IPs/subdomains).") sys.exit(1) # --- Run the scanner --- print(f"[*][Partial Recon] Running VHost & SNI enumeration...") run_vhost_sni_enrichment(recon_data, settings=settings) findings = recon_data.get("vhost_sni", {}).get("findings", []) summary = recon_data.get("vhost_sni", {}).get("summary", {}) print(f"[+][Partial Recon] VHost/SNI scan complete: {summary}") # --- Write findings + link user inputs to the graph --- print(f"[*][Partial Recon] Updating graph database...") try: from graph_db import Neo4jClient with Neo4jClient() as graph_client: if not graph_client.verify_connection(): print("[!][Partial Recon] Neo4j not reachable, graph not updated") return # Ensure user subdomains exist (so HAS_VULNERABILITY can attach) if user_subdomains: driver = graph_client.driver with driver.session() as session: for sub in user_subdomains: session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) ON CREATE SET s.source = 'partial_recon_user_input', s.created_at = datetime() SET s.updated_at = datetime() MERGE (d)-[:HAS_SUBDOMAIN]->(s) """, domain=domain, uid=user_id, pid=project_id, sub=sub, ) for bucket, addrs in (resolved_sub_ips.get(sub) or {}).items(): for addr in addrs or []: session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) MERGE (s)-[:RESOLVES_TO]->(i) """, addr=addr, uid=user_id, pid=project_id, version=bucket, sub=sub, ) # Link user IPs (subdomain attachment OR Generic UserInput) if user_ips: driver = graph_client.driver with driver.session() as session: needs_ip_user_input = False if ip_attach_to: check = session.run( """ MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) RETURN s.name AS name """, sub=ip_attach_to, uid=user_id, pid=project_id, ) if check.single(): for ip_str in user_ips: version = _classify_ip(ip_str) session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (s:Subdomain {name: $sub, user_id: $uid, project_id: $pid}) MERGE (s)-[:RESOLVES_TO]->(i) """, addr=ip_str, uid=user_id, pid=project_id, version=version, sub=ip_attach_to, ) print(f"[+][Partial Recon] Linked {len(user_ips)} IPs to {ip_attach_to}") else: print(f"[!][Partial Recon] Subdomain {ip_attach_to} not found, falling back to UserInput") needs_ip_user_input = True else: needs_ip_user_input = True if needs_ip_user_input: user_input_id = str(uuid.uuid4()) session.run( """ MATCH (d:Domain {name: $domain, user_id: $uid, project_id: $pid}) MERGE (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) ON CREATE SET ui.type = 'ip', ui.values = $ips, ui.created_at = datetime(), ui.tool = 'VhostSni' SET ui.updated_at = datetime() MERGE (d)-[:HAS_USER_INPUT]->(ui) """, domain=domain, uid=user_id, pid=project_id, ui_id=user_input_id, ips=user_ips, ) for ip_str in user_ips: version = _classify_ip(ip_str) session.run( """ MERGE (i:IP {address: $addr, user_id: $uid, project_id: $pid}) ON CREATE SET i.version = $version, i.source = 'partial_recon_user_input', i.created_at = datetime() SET i.updated_at = datetime() WITH i MATCH (ui:UserInput {id: $ui_id, user_id: $uid, project_id: $pid}) MERGE (ui)-[:PRODUCED]->(i) """, addr=ip_str, uid=user_id, pid=project_id, version=version, ui_id=user_input_id, ) print(f"[+][Partial Recon] Created UserInput + linked {len(user_ips)} IPs via PRODUCED") stats = graph_client.update_graph_from_vhost_sni( recon_data=recon_data, user_id=user_id, project_id=project_id, ) print(f"[+][Partial Recon] Graph updated: {json.dumps(stats, default=str)}") except Exception as e: print(f"[!][Partial Recon] Graph update failed: {e}") raise print(f"\n[+][Partial Recon] VHost & SNI enumeration completed successfully ({len(findings)} finding(s))")