"""Cybersecurity pure functions: hashing, parsing, and security utilities.""" import hashlib import math import re import base64 from collections import Counter from urllib.parse import urlparse, urlunparse, parse_qs, urlencode def hash_sha256(data: bytes) -> str: """Calcula el hash SHA-256 de datos binarios. Retorna hex digest.""" return hashlib.sha256(data).hexdigest() def hash_md5(data: bytes) -> str: """Calcula el hash MD5 de datos binarios. Retorna hex digest.""" return hashlib.md5(data).hexdigest() def entropy_shannon(data: bytes) -> float: """Calcula la entropia de Shannon de datos binarios (0-8 bits por byte). Entropia alta (>7.5) sugiere datos cifrados o comprimidos. Entropia baja (<3) sugiere datos estructurados o repetitivos. """ if not data: return 0.0 length = len(data) counts = Counter(data) entropy = 0.0 for count in counts.values(): p = count / length if p > 0: entropy -= p * math.log2(p) return entropy _SQL_INJECTION_PATTERNS = [ (r"('\s*OR\s+'[^']*'\s*=\s*'[^']*'?)", "string_tautology"), (r"('\s*(OR|AND)\s+'?\d+\s*=\s*\d+)", "tautology"), (r"(;\s*(DROP|DELETE|UPDATE|INSERT)\b)", "stacked_query"), (r"(UNION\s+(ALL\s+)?SELECT)", "union_select"), (r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|EXEC)\b\s)", "sql_keyword"), (r"(--\s*$|/\*|\*/)", "comment_injection"), (r"(BENCHMARK\s*\(|SLEEP\s*\(|WAITFOR\s+DELAY)", "time_based"), (r"(CHAR\s*\(\s*\d+)", "char_function"), (r"(CONCAT\s*\()", "concat_function"), (r"(0x[0-9a-fA-F]{4,})", "hex_literal"), ] def detect_sql_injection(input_str: str) -> tuple: """Detecta patrones de SQL injection en un string. Retorna (is_threat, pattern) donde pattern es el nombre del patron detectado o cadena vacia si no hay amenaza. """ for pattern, name in _SQL_INJECTION_PATTERNS: if re.search(pattern, input_str, re.IGNORECASE): return (True, name) return (False, "") _URL_REGEX = re.compile( r"https?://[^\s<>\"'\)\]]+", re.IGNORECASE, ) def extract_urls(text: str) -> list: """Extrae todas las URLs (http/https) de un texto.""" return _URL_REGEX.findall(text) def is_base64(s: str) -> bool: """Verifica si un string es base64 valido. Acepta base64 estandar y URL-safe. Requiere al menos 4 caracteres. """ if len(s) < 4: return False b64_pattern = re.compile(r"^[A-Za-z0-9+/\-_]*={0,2}$") if not b64_pattern.match(s): return False try: decoded = base64.b64decode(s, validate=True) return len(decoded) > 0 except Exception: try: decoded = base64.urlsafe_b64decode(s) return len(decoded) > 0 except Exception: return False def is_hex(s: str) -> bool: """Verifica si un string es hexadecimal valido. Acepta con o sin prefijo 0x. Requiere al menos 2 caracteres (sin prefijo). """ clean = s.strip() if clean.startswith(("0x", "0X")): clean = clean[2:] if len(clean) < 2: return False return bool(re.fullmatch(r"[0-9a-fA-F]+", clean)) def levenshtein_distance(a: str, b: str) -> int: """Calcula la distancia de Levenshtein (edit distance) entre dos strings. Util para deteccion de typosquatting en dominios y fuzzy matching. """ if len(a) < len(b): return levenshtein_distance(b, a) if len(b) == 0: return len(a) prev_row = list(range(len(b) + 1)) for i, ca in enumerate(a): curr_row = [i + 1] for j, cb in enumerate(b): cost = 0 if ca == cb else 1 curr_row.append( min( curr_row[j] + 1, # insert prev_row[j + 1] + 1, # delete prev_row[j] + cost, # substitute ) ) prev_row = curr_row return prev_row[-1] def jaccard_similarity(a: list, b: list) -> float: """Calcula el coeficiente de similitud de Jaccard entre dos listas. J(A,B) = |A interseccion B| / |A union B|. Retorna 0.0 si ambas vacias. Util para comparar conjuntos de tokens, features, o IoCs. """ set_a = set(a) set_b = set(b) if not set_a and not set_b: return 0.0 intersection = set_a & set_b union = set_a | set_b return len(intersection) / len(union) def normalize_url(raw_url: str) -> str: """Normaliza una URL: lowercase del host, elimina fragmentos, ordena parametros. Util para deduplicacion de URLs y comparacion de IoCs. """ parsed = urlparse(raw_url) scheme = parsed.scheme.lower() or "http" netloc = parsed.netloc.lower() path = parsed.path or "/" # Remove trailing slash except for root if path != "/" and path.endswith("/"): path = path.rstrip("/") # Sort query parameters params = parse_qs(parsed.query, keep_blank_values=True) sorted_query = urlencode(sorted(params.items()), doseq=True) # Drop fragment return urlunparse((scheme, netloc, path, parsed.params, sorted_query, ""))