fix(security): pin validated destination IPs for outbound HTTP and TL…

README.md

@@ -594,11 +594,13 @@ To use cloud-hosted serverless PostgreSQL with Neon:
 ## 🛡️ Privacy, Security, and Responsible Use
 
 - **Local-First Boundary**: Secrets (`.env`, encryption keys, SQLite database) are never committed to version control and are ignored by `.gitignore`.
-- **SSRF Defenses**: The audit crawler validates every target domain against DNS rebinding, private IP subnets (`127.0.0.0/8`, `10.0.0.0/8`, `192.168.0.0/16`, `172.16.0.0/12`), and cloud metadata endpoints (`169.254.169.254`).
+- **Connection-Level SSRF Protection & IP Pinning**: Outbound audit probes enforce destination IP pinning at the socket connection level via custom `PinnedAsyncNetworkBackend` (`httpcore`) and `SSRFProtectedTransport` (`httpx`). Target hostnames are resolved and checked against comprehensive prohibited CIDR ranges (RFC 1918 private subnets, RFC 3927 link-local, loopback `127.0.0.0/8`, IPv6 link-local `fe80::/10`, multicast, IPv4-mapped IPv6 `::ffff:0:0/96`, and cloud metadata `169.254.169.254`). The actual TCP connection is pinned to the validated IP, eliminating time-of-check-to-time-of-use (TOCTOU) DNS rebinding attacks while strictly preserving standard TLS Server Name Indication (SNI), TLS certificate hostname verification, and the HTTP `Host` header.
+- **Hop-by-Hop Redirect Revalidation**: Redirects are followed individually (`max_redirects=5`) with each hop's destination re-resolved and security-checked before connecting. Cross-origin sensitive headers (`Authorization`, `Cookie`) are automatically stripped across origins.
+- **Resource and Memory Protection**: Streaming responses are strictly bounded to 5MB (`max_bytes=5242880`) to prevent memory exhaustion from massive bodies or decompression bombs. Transport connections disable keepalive reuse across audit targets (`max_keepalive_connections=0`) to eliminate connection-pool reuse poisoning.
 - **Prompt-Injection Defense**: Model prompts enclose untrusted crawled pages and business names in strict `<untrusted_prospect_data>` XML tags.
-- **Strictly Single-Message Dispatch**: Leddro contains zero mass-mail sequences or automated dispatch triggers. Approving an email draft updates review status in SQLite only; dispatch requires explicit confirmation.
+- **Strictly Single-Message Dispatch**: Leddro contains zero mass-mail sequences or automated dispatch triggers. Approving an email draft updates review status in SQLite only; dispatch requires explicit human confirmation.
 - **Suppression Registry**: Recipient emails and domains are verified against `suppression_records` to prevent messaging unsubscribed contacts.
-- **Single-User Development Disclaimer**: Leddro is built for single-operator local use and does not contain multi-tenant user authentication. Do **not** expose the development server to untrusted public networks without reverse proxy authentication (e.g., HTTP Basic Auth or VPN).
+- **Known Limitations & Operational Boundaries**: Leddro is built for single-operator local use and does not contain multi-tenant user authentication. Probes restrict requests to allowed ports (`80`, `443`, `8080`, `8443`). Do **not** expose the development server to untrusted public networks without reverse proxy authentication (e.g., HTTP Basic Auth or VPN).
 - **Vulnerability Reporting**: See [SECURITY.md](SECURITY.md) for our coordinated vulnerability disclosure policy.
 
 ---
@@ -670,9 +672,12 @@ python -m pytest backend/tests/test_email_scraper.py -v
 # Run Google OAuth & SMTP dispatch test suite
 python -m pytest backend/tests/test_google_oauth_and_email.py -v
 
-# Run SSRF security validation test suite
+# Run SSRF security validation unit test suite (26 tests)
 python -m pytest backend/tests/test_ssrf_validator.py -v
 
+# Run real-socket SSRF connection-pinning integration test suite (8 tests)
+python -m pytest backend/tests/test_ssrf_real_network_integration.py -v
+
 # Run SQLite backup and recovery test suite
 python -m pytest backend/tests/test_backup_and_recovery.py -v
 

backend/app/ai/service.py

@@ -176,13 +176,15 @@ def list_providers(self) -> List[AIProviderInfo]:
 
         # 4. Compatible (Local Ollama / vLLM)
         comp_p = self._providers["compatible"]
+        raw_comp_models = [settings.COMPATIBLE_MODEL, "llama3", "mistral", "qwen2.5"]
+        comp_models = [m.strip() for m in dict.fromkeys(raw_comp_models) if m and m.strip()]
         info_list.append(
             AIProviderInfo(
                 provider_id="compatible",
                 name=comp_p.name,
                 is_configured=comp_p.is_configured(),
                 current_model=settings.COMPATIBLE_MODEL,
-                available_models=[settings.COMPATIBLE_MODEL, "llama3", "mistral", "qwen2.5"],
+                available_models=comp_models,
                 status_message=comp_p.get_status_message(),
                 is_active=(active_id == "compatible"),
                 is_local=True,

backend/app/api/audits.py

@@ -18,7 +18,7 @@
 from backend.app.services.ssrf_validator import validate_target_url
 from backend.app.services.audit_engine import execute_audit
 from backend.app.services.audit_comparison import compare_audits
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, selectinload
 
 router = APIRouter(prefix="/audits", tags=["Audits"])
 
@@ -123,39 +123,48 @@ async def run_audit(payload: AuditRunRequest, db: Session = Depends(get_db)):
 def list_audits(db: Session = Depends(get_db)):
     """
     List all executed audit reports.
+    Uses selectinload to eagerly load findings in a single batched query, eliminating N+1 queries.
     """
-    reports = db.query(AuditReportModel).order_by(AuditReportModel.created_at.desc()).all()
-    results = []
-    for r in reports:
-        findings = db.query(AuditFindingModel).filter(AuditFindingModel.audit_report_id == r.id).all()
-        results.append(db_audit_to_schema(r, findings))
-    return results
+    reports = (
+        db.query(AuditReportModel)
+        .options(selectinload(AuditReportModel.findings))
+        .order_by(AuditReportModel.created_at.desc())
+        .all()
+    )
+    return [db_audit_to_schema(r) for r in reports]
 
 @router.get("/{report_id}")
 def get_audit(report_id: str, db: Session = Depends(get_db)):
     """
     Get audit report details by report ID.
     """
-    report = db.query(AuditReportModel).filter(AuditReportModel.id == report_id).first()
+    report = (
+        db.query(AuditReportModel)
+        .options(selectinload(AuditReportModel.findings))
+        .filter(AuditReportModel.id == report_id)
+        .first()
+    )
     if not report:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Audit report with ID '{report_id}' not found"
         )
-    findings = db.query(AuditFindingModel).filter(AuditFindingModel.audit_report_id == report.id).all()
-    return db_audit_to_schema(report, findings)
+    return db_audit_to_schema(report)
 
 @router.get("/prospect/{prospect_id}")
 def get_prospect_audits(prospect_id: str, db: Session = Depends(get_db)):
     """
     Get all audit reports for a specific prospect.
+    Uses selectinload to eagerly load findings in a single batched query.
     """
-    reports = db.query(AuditReportModel).filter(AuditReportModel.prospect_id == prospect_id).order_by(AuditReportModel.created_at.desc()).all()
-    results = []
-    for r in reports:
-        findings = db.query(AuditFindingModel).filter(AuditFindingModel.audit_report_id == r.id).all()
-        results.append(db_audit_to_schema(r, findings))
-    return results
+    reports = (
+        db.query(AuditReportModel)
+        .options(selectinload(AuditReportModel.findings))
+        .filter(AuditReportModel.prospect_id == prospect_id)
+        .order_by(AuditReportModel.created_at.desc())
+        .all()
+    )
+    return [db_audit_to_schema(r) for r in reports]
 
 @router.post("/compare", response_model=AuditComparisonResponse)
 def compare_two_audits(payload: AuditCompareRequest, db: Session = Depends(get_db)):

backend/app/discovery/deduplication.py

@@ -55,9 +55,10 @@ def check_single_duplicate(
         # Avoid matching generic public hosts
         generic_hosts = {"gmail.com", "yahoo.com", "outlook.com", "facebook.com", "instagram.com", "linkedin.com", "twitter.com", "x.com"}
         if candidate_domain not in generic_hosts:
-            # Query prospects with website_url
+            safe_domain = candidate_domain.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+            # Query only prospects whose website_url contains the candidate domain (eliminates full-table scan)
             prospects_with_url = db.query(ProspectModel).filter(
-                ProspectModel.website_url.isnot(None)
+                ProspectModel.website_url.ilike(f"%{safe_domain}%", escape="\\")
             ).all()
             for p in prospects_with_url:
                 p_domain = extract_registered_domain(p.website_url)
@@ -73,8 +74,14 @@ def check_single_duplicate(
     # 3. Check by Business Name and Location
     norm_candidate_name = normalize_business_name(name)
     if norm_candidate_name and len(norm_candidate_name) >= 3:
-        # Search candidate names
-        potential_matches = db.query(ProspectModel).all()
+        # Efficiently pre-filter candidates by significant name token instead of full-table scan
+        tokens = [t for t in norm_candidate_name.split() if len(t) >= 2]
+        best_token = max(tokens, key=len) if tokens else norm_candidate_name[:3]
+        safe_token = best_token.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+        potential_matches = db.query(ProspectModel).filter(
+            ProspectModel.name.ilike(f"%{safe_token}%", escape="\\")
+        ).all()
         candidate_city = (city or "").strip().lower()
 
         for p in potential_matches: