Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 5,877 Bytes
d543fc1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | import uuid
import hashlib
import hmac
import secrets
import base64
import subprocess
from typing import *
import os
import sys
import re
import json
import time
import urllib3
import requests
import socket
import logging
import threading
import concurrent.futures
import ipaddress
import ssl
from urllib.parse import urlparse, urljoin, urlencode, quote
from collections import defaultdict
from bs4 import BeautifulSoup
from datetime import datetime, timezone
import statistics
"""
anomaly.py — Statistical Anomaly Detectors for Timing & Size Analysis
======================================================================
Used by scanners to detect blind injection via timing differentials.
Improvements (June 2026):
ENH: Minimum 5-sample guard before trusting baseline results.
ENH: build_baseline() accepts proper callable signature.
ENH: z_score() is safe against zero stdev and insufficient samples.
ENH: Added AdaptiveThreshold for dynamic z-score tuning.
"""
class AnomalyDetector:
"""Base statistical detector. Collects numeric samples and detects outliers."""
MIN_BASELINE_SAMPLES = 5 # Require at least this many samples for reliable stats
def __init__(self, baseline_samples: list[float] | None = None):
self._baseline = list(baseline_samples) if baseline_samples else []
def record(self, value: float) -> None:
self._baseline.append(value)
@property
def mean(self) -> float:
return statistics.mean(self._baseline) if self._baseline else 0.0
@property
def stdev(self) -> float:
if len(self._baseline) >= 2:
return statistics.stdev(self._baseline)
return 0.0
@property
def has_baseline(self) -> bool:
return len(self._baseline) >= self.MIN_BASELINE_SAMPLES
def z_score(self, value: float) -> float:
"""
Return z-score of `value` relative to baseline.
Returns 0.0 if stdev is zero or baseline is insufficient.
"""
sd = self.stdev
if sd == 0 or not self._baseline:
return 0.0
return (value - self.mean) / sd
def is_anomalous(self, value: float, threshold: float = 2.5) -> bool:
"""
Returns True only if we have enough baseline AND the z-score exceeds threshold.
Guard against false positives from insufficient data.
"""
if not self.has_baseline:
return False
return abs(self.z_score(value)) >= threshold
def reset(self) -> None:
"""Clear all recorded samples."""
self._baseline.clear()
class TimingAnomalyDetector(AnomalyDetector):
"""
Specialized detector for HTTP response timing analysis.
Used for blind SQLi, CMDi, SSTI, SSRF timing-based detection.
"""
def __init__(self, baseline_samples: list[float] | None = None):
super().__init__(baseline_samples)
self._timing_records: list[tuple[str, float, str]] = []
def record_timing(self, label: str, elapsed: float, payload: str = "") -> None:
self.record(elapsed)
self._timing_records.append((label, elapsed, payload))
def build_baseline(
self,
request_fn,
url: str,
n: int = 5,
headers: dict | None = None,
method: str = "GET",
) -> None:
"""
Build timing baseline by making `n` requests to `url`.
FIX: Accepts request_fn with signature (url, method, data, headers, timeout).
Passes all positional args to avoid keyword-mismatch errors when
callers pass `self._make_request` directly.
"""
for _ in range(n):
t0 = time.monotonic()
try:
# Use positional args to match BaseScanner._make_request signature:
# (url, method="GET", data=None, headers=None, timeout=8)
request_fn(url, method, None, headers or {}, 8)
except Exception:
pass # Network errors are expected during baseline
self.record(time.monotonic() - t0)
def test_payload(
self,
label: str,
elapsed: float,
payload: str = "",
z_threshold: float = 3.0,
) -> bool:
"""
Record a timed payload request and test if it's anomalous.
Returns True if response time is statistically abnormal.
"""
self.record_timing(label, elapsed, payload)
return self.is_anomalous(elapsed, z_threshold)
class SizeAnomalyDetector(AnomalyDetector):
"""
Specialized detector for HTTP response size analysis.
Used for boolean-based blind injection (different sizes for true/false conditions).
"""
MIN_BASELINE_SAMPLES = 3 # Size detection can work with fewer samples
def __init__(self, baseline_sizes: list[int] | None = None):
sizes = [float(s) for s in (baseline_sizes or [])]
super().__init__(sizes)
def record_size(self, size: int) -> None:
self.record(float(size))
def test_size(self, size: int, z_threshold: float = 2.5) -> bool:
return self.is_anomalous(float(size), z_threshold)
def seed_pair(self, true_len: int, false_len: int) -> None:
"""
Seed with true/false response sizes to initialize comparison.
Adds both as baseline samples.
"""
self.record_size(true_len)
self.record_size(false_len)
def pair_differs(self, true_len: int, false_len: int, min_diff: int = 30) -> bool:
"""
Simple heuristic: return True if the two response sizes differ
by at least `min_diff` bytes — used before enough baseline exists.
"""
return abs(true_len - false_len) >= min_diff
|