From 984c9ade6ca7bc3e824b76fd03d8e11c7639cb49 Mon Sep 17 00:00:00 2001 From: Aleksei Fedorov Date: Fri, 3 Apr 2026 19:16:24 +0000 Subject: [PATCH] python refactoring 5 (crypto security) --- kernel_ai/services/crypto_security.py | 436 +++++++++++++++++++-- kernel_ai/webapp.py | 521 +------------------------- 2 files changed, 411 insertions(+), 546 deletions(-) diff --git a/kernel_ai/services/crypto_security.py b/kernel_ai/services/crypto_security.py index 64ea5fc..5c11ae9 100644 --- a/kernel_ai/services/crypto_security.py +++ b/kernel_ai/services/crypto_security.py @@ -11,22 +11,406 @@ import psutil -def collect_crypto_realtime(callbacks, crypto_prev): +def infer_crypto_protocol(local_port, remote_port, process_name): + """Infer protocol/algorithm pair from ports and process hints.""" + ports = {int(local_port or 0), int(remote_port or 0)} + p_name = (process_name or "").lower() + if 22 in ports or "ssh" in p_name: + return "SSH", "Curve25519/ChaCha20-Poly1305" + if 51820 in ports or "wireguard" in p_name or p_name.startswith("wg"): + return "WireGuard", "ChaCha20-Poly1305" + tls_ports = {443, 465, 636, 853, 993, 995, 8443, 9443, 6443, 2376} + if ports & tls_ports or any(x in p_name for x in ["nginx", "haproxy", "curl", "wget", "openssl", "stunnel", "traefik"]): + return "TLS", "AES-GCM/SHA256" + return "Crypto API", "AES/SHA" + + +def is_likely_crypto_actor(process_name, local_port, remote_port, protocol): + """Heuristic gate to avoid flooding with unrelated sockets.""" + p_name = (process_name or "").lower() + interesting_ports = {22, 443, 465, 636, 853, 993, 995, 2376, 6443, 8443, 9443, 51820} + process_tokens = [ + "nginx", + "haproxy", + "envoy", + "caddy", + "apache", + "httpd", + "traefik", + "sshd", + "ssh", + "wg", + "wireguard", + "openssl", + "stunnel", + "curl", + "wget", + "python", + "gunicorn", + "uvicorn", + ] + if protocol in ("TLS", "SSH", "WireGuard"): + return True + if int(local_port or 0) in interesting_ports or int(remote_port or 0) in interesting_ports: + return True + return any(token in p_name for token in process_tokens) + + +def infer_tls_terminator(process_name, local_port, protocol, tls_listener_names): + """Guess where TLS termination happens.""" + if protocol != "TLS": + return "n/a" + p_name = (process_name or "").lower() + if p_name and p_name != "unknown": + return p_name + if int(local_port or 0) in {443, 8443, 9443, 6443} and tls_listener_names: + top = next(iter(tls_listener_names)) + return f"listener:{top}" + if int(local_port or 0) in {443, 8443, 9443, 6443}: + return "unknown" + return "upstream-or-external-lb" + + +def parse_proc_crypto_entries(): + """Parse /proc/crypto into a list of dict entries.""" + entries = [] + try: + with open("/proc/crypto", "r", encoding="utf-8", errors="ignore") as f: + raw = f.read() + except Exception: + return entries + + blocks = [block.strip() for block in raw.split("\n\n") if block.strip()] + for block in blocks: + item = {} + for line in block.splitlines(): + if ":" not in line: + continue + key, value = line.split(":", 1) + item[key.strip().lower()] = value.strip() + if item: + entries.append(item) + return entries + + +def collect_algorithm_competition(requested_algorithm="aes"): + """Build algorithm competition using kernel crypto registry.""" + entries = parse_proc_crypto_entries() + requested = (requested_algorithm or "aes").lower() + req_type_allow = {"aes": {"skcipher", "aead", "cipher"}, "sha": {"shash", "ahash", "hash"}, "chacha20": {"skcipher", "aead", "cipher"}} + req_tokens = {"aes": ["aes"], "sha": ["sha"], "chacha20": ["chacha20", "xchacha20", "chacha"]} + allowed_types = req_type_allow.get(requested, {"skcipher", "aead", "cipher", "shash", "ahash", "hash"}) + tokens = req_tokens.get(requested, [requested]) + candidates = [] + + for entry in entries: + name = str(entry.get("name", "")).lower() + driver = str(entry.get("driver", "")).lower() + alg_type = str(entry.get("type", "")).lower() + if not any(token in name or token in driver for token in tokens): + continue + if alg_type and alg_type not in allowed_types: + continue + try: + priority = int(entry.get("priority", "0") or 0) + except ValueError: + priority = 0 + impl_name = driver or name or "unknown-impl" + candidates.append({"name": impl_name, "priority": priority, "type": alg_type or "unknown", "source": "kernel"}) + + dedup = {} + for item in candidates: + existing = dedup.get(item["name"]) + if existing is None or item["priority"] > existing["priority"]: + dedup[item["name"]] = item + candidates = list(dedup.values()) + candidates.sort(key=lambda x: x["priority"], reverse=True) + + if not candidates: + fallback_map = { + "aes": [{"name": "aesni-intel", "priority": 300, "type": "skcipher", "source": "mock"}, {"name": "aes-avx", "priority": 200, "type": "skcipher", "source": "mock"}, {"name": "aes-generic", "priority": 100, "type": "skcipher", "source": "mock"}], + "sha": [{"name": "sha256-avx2", "priority": 240, "type": "shash", "source": "mock"}, {"name": "sha256-ssse3", "priority": 180, "type": "shash", "source": "mock"}, {"name": "sha256-generic", "priority": 100, "type": "shash", "source": "mock"}], + "chacha20": [{"name": "chacha20-neon", "priority": 260, "type": "skcipher", "source": "mock"}, {"name": "chacha20-simd", "priority": 220, "type": "skcipher", "source": "mock"}, {"name": "chacha20-generic", "priority": 100, "type": "skcipher", "source": "mock"}], + } + candidates = fallback_map.get(requested, fallback_map["aes"]) + + selected = candidates[0] if candidates else None + return {"request": requested.upper(), "implementations": candidates[:8], "selected": selected, "selection_policy": "max-priority"} + + +def collect_kernel_crypto_clients(items): + """Infer major kernel crypto clients from active context.""" + client_rules = [ + ("kTLS", ["nginx", "haproxy", "envoy", "caddy", "apache", "httpd", "traefik"], "TLS"), + ("WireGuard", ["wg", "wireguard"], "WireGuard"), + ("IPsec/XFRM", ["charon", "strongswan", "ipsec", "racoon"], "TLS"), + ("dm-crypt", ["cryptsetup", "dmcrypt", "luks"], "CRYPTO API"), + ("fscrypt", ["fscrypt"], "CRYPTO API"), + ("AF_ALG", ["openssl", "python", "curl", "wget"], "CRYPTO API"), + ] + results = [] + lowered_items = [{"process": str(item.get("process", "")).lower(), "protocol": str(item.get("protocol", ""))} for item in items] + for name, tokens, proto_hint in client_rules: + flows = 0 + for item in lowered_items: + proc = item["process"] + proto = item["protocol"] + if any(token in proc for token in tokens): + flows += 1 + elif proto_hint and proto == proto_hint: + flows += 1 + status = "active" if flows > 0 else "idle" + results.append({"name": name, "status": status, "active_flows": int(flows)}) + return results + + +def collect_sync_async_queue(items): + """Estimate sync/async crypto execution pressure from active flows.""" + active_items = [i for i in items if str(i.get("status", "")).upper() != "LISTEN"] + async_items = [i for i in active_items if str(i.get("source_kind", "")) == "connection" or str(i.get("protocol", "")).upper() in {"TLS", "WIREGUARD", "SSH"}] + sync_items = max(len(active_items) - len(async_items), 0) + sum(1 for i in items if str(i.get("source_kind", "")) == "process") + queue_depth = max(len(async_items) - 1, 0) + queue_latency_ms = round(0.35 + min(5.5, queue_depth * 0.42 + len(active_items) * 0.08), 2) + return {"sync_ops_est": int(sync_items), "async_ops_est": int(len(async_items)), "queue_depth_est": int(queue_depth), "queue_latency_ms_est": queue_latency_ms, "mode": "heuristic"} + + +def collect_hw_offload_status(entries, algorithm_competitions): + """Estimate hardware acceleration availability from /proc/crypto drivers.""" + names = [] + for entry in entries: + n = str(entry.get("name", "")).lower() + d = str(entry.get("driver", "")).lower() + if n: + names.append(n) + if d: + names.append(d) + + def has_token(tokens): + return any(any(token in item for token in tokens) for item in names) + + selected_impls = {key: str(value.get("selected", {}).get("name", "")).lower() for key, value in (algorithm_competitions or {}).items()} + selected_joined = " ".join(selected_impls.values()) + engines = [ + {"engine": "AES-NI / CPU INSTR", "available": has_token(["aesni", "vaes"]), "active": ("aesni" in selected_joined or "vaes" in selected_joined)}, + {"engine": "SIMD (AVX/NEON)", "available": has_token(["avx", "sse", "simd", "neon"]), "active": any(token in selected_joined for token in ["avx", "simd", "neon", "sse"])}, + {"engine": "ARM CRYPTO EXT", "available": has_token(["arm64", "ce", "neon"]), "active": "arm64" in selected_joined}, + {"engine": "QAT OFFLOAD", "available": has_token(["qat"]), "active": "qat" in selected_joined}, + {"engine": "VIRTIO-CRYPTO", "available": has_token(["virtio"]), "active": "virtio" in selected_joined}, + ] + result = [] + for item in engines: + if item["active"]: + status = "active" + elif item["available"]: + status = "available" + else: + status = "unavailable" + result.append({"engine": item["engine"], "status": status}) + return result + + +def read_sysctl_int(path, default=0): + try: + with open(path, "r", encoding="utf-8", errors="ignore") as f: + raw = f.read().strip() + return int(raw or default) + except Exception: + return int(default) + + +def read_proc_interrupt_total(): + try: + with open("/proc/stat", "r", encoding="utf-8", errors="ignore") as f: + for line in f: + if line.startswith("intr "): + parts = line.strip().split() + if len(parts) >= 2: + return int(parts[1]) + except Exception: + return 0 + return 0 + + +def collect_entropy_cloud_status(entropy_prev): + """Collect Linux random subsystem entropy status and source activity.""" + now = time.time() + entropy_bits = read_sysctl_int("/proc/sys/kernel/random/entropy_avail", 0) + pool_size_bits = read_sysctl_int("/proc/sys/kernel/random/poolsize", 256) + read_threshold = read_sysctl_int("/proc/sys/kernel/random/read_wakeup_threshold", 128) + write_threshold = read_sysctl_int("/proc/sys/kernel/random/write_wakeup_threshold", 64) + try: + disk = psutil.disk_io_counters() + except Exception: + disk = None + try: + net = psutil.net_io_counters() + except Exception: + net = None + intr_total = read_proc_interrupt_total() + prev_ts = entropy_prev.get("timestamp") + dt = max(now - prev_ts, 0.001) if prev_ts else None + disk_read_now = int(getattr(disk, "read_bytes", 0) or 0) + disk_write_now = int(getattr(disk, "write_bytes", 0) or 0) + net_sent_now = int(getattr(net, "bytes_sent", 0) or 0) + net_recv_now = int(getattr(net, "bytes_recv", 0) or 0) + if dt: + disk_delta = max((disk_read_now - int(entropy_prev.get("disk_read_bytes") or disk_read_now)) + (disk_write_now - int(entropy_prev.get("disk_write_bytes") or disk_write_now)), 0) + net_delta = max((net_sent_now - int(entropy_prev.get("net_sent_bytes") or net_sent_now)) + (net_recv_now - int(entropy_prev.get("net_recv_bytes") or net_recv_now)), 0) + intr_delta = max(intr_total - int(entropy_prev.get("interrupt_total") or intr_total), 0) + else: + disk_delta = 0 + net_delta = 0 + intr_delta = 0 + entropy_prev["timestamp"] = now + entropy_prev["disk_read_bytes"] = disk_read_now + entropy_prev["disk_write_bytes"] = disk_write_now + entropy_prev["net_sent_bytes"] = net_sent_now + entropy_prev["net_recv_bytes"] = net_recv_now + entropy_prev["interrupt_total"] = intr_total + + def scale_intensity(rate_value, scale): + return int(max(0, min(100, (float(rate_value) / float(scale)) * 100.0))) + + disk_rate = (disk_delta / dt) if dt else 0 + net_rate = (net_delta / dt) if dt else 0 + intr_rate = (intr_delta / dt) if dt else 0 + irq_intensity = scale_intensity(intr_rate, 25000) + disk_intensity = scale_intensity(disk_rate, 80 * 1024 * 1024) + net_intensity = scale_intensity(net_rate, 120 * 1024 * 1024) + hwrng_intensity = 68 if entropy_bits > max(read_threshold, 128) else 34 + sources = [ + {"source": "interrupt timing", "intensity": irq_intensity, "status": "active" if irq_intensity >= 25 else "low"}, + {"source": "disk IO", "intensity": disk_intensity, "status": "active" if disk_intensity >= 18 else "low"}, + {"source": "network timing", "intensity": net_intensity, "status": "active" if net_intensity >= 18 else "low"}, + {"source": "hardware RNG", "intensity": hwrng_intensity, "status": "active" if hwrng_intensity >= 50 else "limited"}, + ] + source_avg = int(sum(s["intensity"] for s in sources) / max(len(sources), 1)) + entropy_pct = max(0.0, min(1.0, float(entropy_bits) / max(float(pool_size_bits), 1.0))) + particle_density = max(16, min(84, int(18 + entropy_pct * 42 + source_avg * 0.35))) + key_birth_rate = round(0.6 + entropy_pct * 9.4 + source_avg * 0.06, 2) + crng_state = "ready" if entropy_bits >= max(read_threshold, 128) else "warming" + random_state = "stable" if entropy_bits >= max(write_threshold, 64) else "refilling" + return { + "entropy_pool_bits": int(entropy_bits), + "entropy_pool_size_bits": int(pool_size_bits), + "crng_state": crng_state, + "random_subsystem_state": random_state, + "particle_density": int(particle_density), + "key_birth_rate_est": float(key_birth_rate), + "sources": sources, + "read_wakeup_threshold": int(read_threshold), + "write_wakeup_threshold": int(write_threshold), + "mode": "live-heuristic", + } + + +def collect_algorithm_requesters(items, kernel_clients): + """Infer likely requestor objects that trigger algorithm competition.""" + algo_map = {"aes": {}, "sha": {}, "chacha20": {}} + client_boost_rules = {"aes": {"kTLS", "dm-crypt", "AF_ALG", "IPsec/XFRM"}, "sha": {"kTLS", "AF_ALG", "IPsec/XFRM"}, "chacha20": {"WireGuard", "AF_ALG"}} + for item in items or []: + process_name = str(item.get("process", "unknown")).lower() or "unknown" + protocol = str(item.get("protocol", "")).upper() + algorithm = str(item.get("algorithm", "")).upper() + status = str(item.get("status", "")).upper() + if status == "LISTEN": + continue + matched_algorithms = set() + if "AES" in algorithm or protocol == "TLS": + matched_algorithms.add("aes") + if "SHA" in algorithm or protocol == "TLS": + matched_algorithms.add("sha") + if "CHACHA" in algorithm or protocol in {"WIREGUARD", "SSH"}: + matched_algorithms.add("chacha20") + if not matched_algorithms and protocol == "CRYPTO API": + matched_algorithms.update(["aes", "sha"]) + for algo_key in matched_algorithms: + key = f"process:{process_name}" + bucket = algo_map[algo_key].setdefault(key, {"name": process_name, "kind": "process", "score": 0}) + bucket["score"] += 1 + for client in kernel_clients or []: + name = str(client.get("name", "")).strip() + flows = int(client.get("active_flows", 0) or 0) + if not name or flows <= 0: + continue + for algo_key, allowed_clients in client_boost_rules.items(): + if name not in allowed_clients: + continue + key = f"client:{name}" + bucket = algo_map[algo_key].setdefault(key, {"name": name, "kind": "kernel-client", "score": 0}) + bucket["score"] += max(2, flows) + result = {} + for algo_key, raw in algo_map.items(): + ranked = sorted(raw.values(), key=lambda x: x.get("score", 0), reverse=True) + if not ranked: + ranked = [{"name": "user/kernel request", "kind": "generic", "score": 1}] + result[algo_key] = ranked[:4] + return result + + +def build_crypto_decision_pipelines(algorithm_competitions, kernel_clients, hw_offload, algorithm_requesters): + """Build visual decision pipeline metadata for each algorithm family.""" + hw_active = [h.get("engine") for h in (hw_offload or []) if h.get("status") == "active"] + hw_available = [h.get("engine") for h in (hw_offload or []) if h.get("status") == "available"] + capability_hint = ", ".join(hw_active[:2] or hw_available[:2]) if (hw_active or hw_available) else "generic-cpu-only" + tfm_lookup_map = {"AES": "crypto_alloc_skcipher(aes)", "SHA": "crypto_alloc_shash(sha*)", "CHACHA20": "crypto_alloc_skcipher(chacha20)"} + pipelines = {} + for key, comp in (algorithm_competitions or {}).items(): + request = str(comp.get("request", key)).upper() + impls = comp.get("implementations", []) or [] + shortlist = [str(x.get("name", "unknown")) for x in impls[:3]] + requesters = list((algorithm_requesters or {}).get(key, [])) + top_requester = requesters[0] if requesters else {"name": "user/kernel request", "kind": "generic"} + request_origin = f"{top_requester.get('kind', 'generic')}: {top_requester.get('name', 'unknown')}" + selected_driver = str((comp.get("selected") or {}).get("name", "unknown")) + fallback_driver = next((name for name in shortlist if "generic" in name.lower()), shortlist[-1] if shortlist else "none") + selected_is_generic = "generic" in selected_driver.lower() + selected_source = str((comp.get("selected") or {}).get("source", "kernel")).lower() + fallback_active = selected_is_generic and len(shortlist) > 1 + pipelines[key] = { + "request": request, + "request_origin": request_origin, + "requesters": requesters, + "tfm_lookup": tfm_lookup_map.get(request, f"crypto_lookup({request.lower()})"), + "impl_shortlist": shortlist, + "priority_check": "max priority wins", + "capability_check": capability_hint, + "selected_driver": selected_driver, + "fallback_driver": fallback_driver, + "fallback_active": bool(fallback_active), + "fallback_reason": "higher-priority impl unavailable or unsupported" if fallback_active else "not-triggered", + "source": selected_source, + } + return pipelines + + +def collect_crypto_realtime(crypto_prev, entropy_prev=None, callbacks=None): """ Build a near-realtime list of processes likely interacting with kernel crypto. This is heuristic-based and derived from active network/process context. """ - infer_crypto_protocol = callbacks["infer_crypto_protocol"] - is_likely_crypto_actor = callbacks["is_likely_crypto_actor"] - infer_tls_terminator = callbacks["infer_tls_terminator"] - collect_algorithm_competition = callbacks["collect_algorithm_competition"] - parse_proc_crypto_entries = callbacks["parse_proc_crypto_entries"] - collect_kernel_crypto_clients = callbacks["collect_kernel_crypto_clients"] - collect_hw_offload_status = callbacks["collect_hw_offload_status"] - collect_sync_async_queue = callbacks["collect_sync_async_queue"] - collect_algorithm_requesters = callbacks["collect_algorithm_requesters"] - build_crypto_decision_pipelines = callbacks["build_crypto_decision_pipelines"] - collect_entropy_cloud_status = callbacks["collect_entropy_cloud_status"] + callbacks = callbacks or {} + infer_crypto_protocol_fn = callbacks.get("infer_crypto_protocol", infer_crypto_protocol) + is_likely_crypto_actor_fn = callbacks.get("is_likely_crypto_actor", is_likely_crypto_actor) + infer_tls_terminator_fn = callbacks.get("infer_tls_terminator", infer_tls_terminator) + collect_algorithm_competition_fn = callbacks.get("collect_algorithm_competition", collect_algorithm_competition) + parse_proc_crypto_entries_fn = callbacks.get("parse_proc_crypto_entries", parse_proc_crypto_entries) + collect_kernel_crypto_clients_fn = callbacks.get("collect_kernel_crypto_clients", collect_kernel_crypto_clients) + collect_hw_offload_status_fn = callbacks.get("collect_hw_offload_status", collect_hw_offload_status) + collect_sync_async_queue_fn = callbacks.get("collect_sync_async_queue", collect_sync_async_queue) + collect_algorithm_requesters_fn = callbacks.get("collect_algorithm_requesters", collect_algorithm_requesters) + build_crypto_decision_pipelines_fn = callbacks.get("build_crypto_decision_pipelines", build_crypto_decision_pipelines) + collect_entropy_cloud_status_fn = callbacks.get("collect_entropy_cloud_status") + entropy_prev_local = entropy_prev or { + "timestamp": None, + "disk_read_bytes": 0, + "disk_write_bytes": 0, + "net_sent_bytes": 0, + "net_recv_bytes": 0, + "interrupt_total": 0, + } + if collect_entropy_cloud_status_fn is None: + collect_entropy_cloud_status_fn = lambda: collect_entropy_cloud_status(entropy_prev_local) items = [] tls_listener_by_port = {} @@ -99,11 +483,11 @@ def collect_crypto_realtime(callbacks, crypto_prev): if listener_meta: process_name = listener_meta.get("process") or "unknown" - protocol, algorithm = infer_crypto_protocol(local_port, remote_port, process_name) - if not is_likely_crypto_actor(process_name, local_port, remote_port, protocol): + protocol, algorithm = infer_crypto_protocol_fn(local_port, remote_port, process_name) + if not is_likely_crypto_actor_fn(process_name, local_port, remote_port, protocol): continue - tls_terminator = infer_tls_terminator(process_name, local_port, protocol, tls_listener_names) + tls_terminator = infer_tls_terminator_fn(process_name, local_port, protocol, tls_listener_names) endpoint = f"{remote_ip}:{remote_port}" if remote_ip else f"{local_ip}:{local_port}" items.append( @@ -128,7 +512,7 @@ def collect_crypto_realtime(callbacks, crypto_prev): except Exception: continue if any(token in name for token in ["nginx", "sshd", "curl", "openssl", "kube", "vpn", "python"]): - protocol, algorithm = infer_crypto_protocol(0, 0, name) + protocol, algorithm = infer_crypto_protocol_fn(0, 0, name) items.append( { "process": name, @@ -182,26 +566,26 @@ def collect_crypto_realtime(callbacks, crypto_prev): unique_processes.append(p) algorithm_competitions = { - "aes": collect_algorithm_competition("aes"), - "sha": collect_algorithm_competition("sha"), - "chacha20": collect_algorithm_competition("chacha20"), + "aes": collect_algorithm_competition_fn("aes"), + "sha": collect_algorithm_competition_fn("sha"), + "chacha20": collect_algorithm_competition_fn("chacha20"), } - proc_crypto_entries = parse_proc_crypto_entries() - kernel_clients = collect_kernel_crypto_clients(items) - hw_offload = collect_hw_offload_status(proc_crypto_entries, algorithm_competitions) + proc_crypto_entries = parse_proc_crypto_entries_fn() + kernel_clients = collect_kernel_crypto_clients_fn(items) + hw_offload = collect_hw_offload_status_fn(proc_crypto_entries, algorithm_competitions) crypto_stage1 = { "kernel_clients": kernel_clients, - "sync_async": collect_sync_async_queue(items), + "sync_async": collect_sync_async_queue_fn(items), "hw_offload": hw_offload, } - algorithm_requesters = collect_algorithm_requesters(items, kernel_clients) - crypto_decision_pipelines = build_crypto_decision_pipelines( + algorithm_requesters = collect_algorithm_requesters_fn(items, kernel_clients) + crypto_decision_pipelines = build_crypto_decision_pipelines_fn( algorithm_competitions=algorithm_competitions, kernel_clients=kernel_clients, hw_offload=hw_offload, algorithm_requesters=algorithm_requesters, ) - entropy_cloud = collect_entropy_cloud_status() + entropy_cloud = collect_entropy_cloud_status_fn() return { "items": items[:24], diff --git a/kernel_ai/webapp.py b/kernel_ai/webapp.py index 00f583d..fe30282 100644 --- a/kernel_ai/webapp.py +++ b/kernel_ai/webapp.py @@ -1485,527 +1485,8 @@ def map_interrupt_to_subsystem(interrupt_name): else: return 'kernel' -def infer_crypto_protocol(local_port, remote_port, process_name): - """Infer protocol likely using kernel crypto from socket and process context.""" - tls_ports = {443, 465, 636, 853, 993, 995, 2376, 6443, 8443, 9443} - ssh_ports = {22} - wg_ports = {51820} - p_name = (process_name or "").lower() - ports = {int(local_port or 0), int(remote_port or 0)} - - if ports & ssh_ports or "sshd" in p_name or "ssh" in p_name: - return "SSH", "ChaCha20-Poly1305" - if ports & wg_ports or "wg" in p_name or "wireguard" in p_name: - return "WireGuard", "ChaCha20" - if ports & tls_ports or any(x in p_name for x in ["nginx", "haproxy", "curl", "wget", "openssl", "stunnel", "traefik"]): - return "TLS", "AES-GCM/SHA256" - return "Crypto API", "AES/SHA" - -def is_likely_crypto_actor(process_name, local_port, remote_port, protocol): - """Heuristic gate to avoid flooding with unrelated sockets.""" - p_name = (process_name or "").lower() - interesting_ports = {22, 443, 465, 636, 853, 993, 995, 2376, 6443, 8443, 9443, 51820} - process_tokens = [ - "nginx", "haproxy", "envoy", "caddy", "apache", "httpd", "traefik", - "sshd", "ssh", "wg", "wireguard", "openssl", "stunnel", "curl", "wget", - "python", "gunicorn", "uvicorn" - ] - if protocol in ("TLS", "SSH", "WireGuard"): - return True - if int(local_port or 0) in interesting_ports or int(remote_port or 0) in interesting_ports: - return True - return any(token in p_name for token in process_tokens) - -def infer_tls_terminator(process_name, local_port, protocol, tls_listener_names): - """Guess where TLS termination happens.""" - if protocol != "TLS": - return "n/a" - p_name = (process_name or "").lower() - if p_name and p_name != "unknown": - return p_name - if int(local_port or 0) in {443, 8443, 9443, 6443} and tls_listener_names: - top = next(iter(tls_listener_names)) - return f"listener:{top}" - if int(local_port or 0) in {443, 8443, 9443, 6443}: - return "unknown" - return "upstream-or-external-lb" - -def parse_proc_crypto_entries(): - """Parse /proc/crypto into a list of dict entries.""" - entries = [] - try: - with open("/proc/crypto", "r", encoding="utf-8", errors="ignore") as f: - raw = f.read() - except Exception: - return entries - - blocks = [block.strip() for block in raw.split("\n\n") if block.strip()] - for block in blocks: - item = {} - for line in block.splitlines(): - if ":" not in line: - continue - key, value = line.split(":", 1) - item[key.strip().lower()] = value.strip() - if item: - entries.append(item) - return entries - -def collect_algorithm_competition(requested_algorithm="aes"): - """ - Build algorithm implementation competition using kernel crypto registry. - The winner is the implementation with highest priority. - """ - entries = parse_proc_crypto_entries() - requested = (requested_algorithm or "aes").lower() - req_type_allow = { - "aes": {"skcipher", "aead", "cipher"}, - "sha": {"shash", "ahash", "hash"}, - "chacha20": {"skcipher", "aead", "cipher"} - } - req_tokens = { - "aes": ["aes"], - "sha": ["sha"], - "chacha20": ["chacha20", "xchacha20", "chacha"] - } - allowed_types = req_type_allow.get(requested, {"skcipher", "aead", "cipher", "shash", "ahash", "hash"}) - tokens = req_tokens.get(requested, [requested]) - candidates = [] - - for entry in entries: - name = str(entry.get("name", "")).lower() - driver = str(entry.get("driver", "")).lower() - alg_type = str(entry.get("type", "")).lower() - - if not any(token in name or token in driver for token in tokens): - continue - if alg_type and alg_type not in allowed_types: - continue - - try: - priority = int(entry.get("priority", "0") or 0) - except ValueError: - priority = 0 - - impl_name = driver or name or "unknown-impl" - candidates.append({ - "name": impl_name, - "priority": priority, - "type": alg_type or "unknown", - "source": "kernel" - }) - - # Deduplicate by implementation name, keep the highest priority variant. - dedup = {} - for item in candidates: - existing = dedup.get(item["name"]) - if existing is None or item["priority"] > existing["priority"]: - dedup[item["name"]] = item - candidates = list(dedup.values()) - candidates.sort(key=lambda x: x["priority"], reverse=True) - - if not candidates: - # Fallback keeps the UX informative on hosts without readable /proc/crypto. - fallback_map = { - "aes": [ - {"name": "aesni-intel", "priority": 300, "type": "skcipher", "source": "mock"}, - {"name": "aes-avx", "priority": 200, "type": "skcipher", "source": "mock"}, - {"name": "aes-generic", "priority": 100, "type": "skcipher", "source": "mock"} - ], - "sha": [ - {"name": "sha256-avx2", "priority": 240, "type": "shash", "source": "mock"}, - {"name": "sha256-ssse3", "priority": 180, "type": "shash", "source": "mock"}, - {"name": "sha256-generic", "priority": 100, "type": "shash", "source": "mock"} - ], - "chacha20": [ - {"name": "chacha20-neon", "priority": 260, "type": "skcipher", "source": "mock"}, - {"name": "chacha20-simd", "priority": 220, "type": "skcipher", "source": "mock"}, - {"name": "chacha20-generic", "priority": 100, "type": "skcipher", "source": "mock"} - ] - } - candidates = fallback_map.get(requested, fallback_map["aes"]) - - selected = candidates[0] if candidates else None - return { - "request": requested.upper(), - "implementations": candidates[:8], - "selected": selected, - "selection_policy": "max-priority" - } - -def collect_kernel_crypto_clients(items): - """Infer major kernel crypto clients from active process/protocol context.""" - client_rules = [ - ("kTLS", ["nginx", "haproxy", "envoy", "caddy", "apache", "httpd", "traefik"], "TLS"), - ("WireGuard", ["wg", "wireguard"], "WireGuard"), - ("IPsec/XFRM", ["charon", "strongswan", "ipsec", "racoon"], "TLS"), - ("dm-crypt", ["cryptsetup", "dmcrypt", "luks"], "CRYPTO API"), - ("fscrypt", ["fscrypt"], "CRYPTO API"), - ("AF_ALG", ["openssl", "python", "curl", "wget"], "CRYPTO API") - ] - results = [] - lowered_items = [] - for item in items: - lowered_items.append({ - "process": str(item.get("process", "")).lower(), - "protocol": str(item.get("protocol", "")), - "source_kind": str(item.get("source_kind", "")) - }) - - for name, tokens, proto_hint in client_rules: - flows = 0 - for item in lowered_items: - proc = item["process"] - proto = item["protocol"] - if any(token in proc for token in tokens): - flows += 1 - elif proto_hint and proto == proto_hint: - flows += 1 - status = "active" if flows > 0 else "idle" - results.append({ - "name": name, - "status": status, - "active_flows": int(flows) - }) - return results - -def collect_sync_async_queue(items): - """Estimate sync/async crypto execution pressure from active flows.""" - active_items = [i for i in items if str(i.get("status", "")).upper() != "LISTEN"] - async_items = [ - i for i in active_items - if str(i.get("source_kind", "")) == "connection" - or str(i.get("protocol", "")).upper() in {"TLS", "WIREGUARD", "SSH"} - ] - sync_items = max(len(active_items) - len(async_items), 0) + sum( - 1 for i in items if str(i.get("source_kind", "")) == "process" - ) - queue_depth = max(len(async_items) - 1, 0) - queue_latency_ms = round(0.35 + min(5.5, queue_depth * 0.42 + len(active_items) * 0.08), 2) - return { - "sync_ops_est": int(sync_items), - "async_ops_est": int(len(async_items)), - "queue_depth_est": int(queue_depth), - "queue_latency_ms_est": queue_latency_ms, - "mode": "heuristic" - } - -def collect_hw_offload_status(entries, algorithm_competitions): - """Estimate hardware acceleration availability from /proc/crypto drivers.""" - names = [] - for entry in entries: - n = str(entry.get("name", "")).lower() - d = str(entry.get("driver", "")).lower() - if n: - names.append(n) - if d: - names.append(d) - - def has_token(tokens): - return any(any(token in item for token in tokens) for item in names) - - selected_impls = { - key: str(value.get("selected", {}).get("name", "")).lower() - for key, value in (algorithm_competitions or {}).items() - } - selected_joined = " ".join(selected_impls.values()) - - engines = [ - { - "engine": "AES-NI / CPU INSTR", - "available": has_token(["aesni", "vaes"]), - "active": ("aesni" in selected_joined or "vaes" in selected_joined) - }, - { - "engine": "SIMD (AVX/NEON)", - "available": has_token(["avx", "sse", "simd", "neon"]), - "active": any(token in selected_joined for token in ["avx", "simd", "neon", "sse"]) - }, - { - "engine": "ARM CRYPTO EXT", - "available": has_token(["arm64", "ce", "neon"]), - "active": "arm64" in selected_joined - }, - { - "engine": "QAT OFFLOAD", - "available": has_token(["qat"]), - "active": "qat" in selected_joined - }, - { - "engine": "VIRTIO-CRYPTO", - "available": has_token(["virtio"]), - "active": "virtio" in selected_joined - } - ] - result = [] - for item in engines: - if item["active"]: - status = "active" - elif item["available"]: - status = "available" - else: - status = "unavailable" - result.append({ - "engine": item["engine"], - "status": status - }) - return result - -def read_sysctl_int(path, default=0): - """Read integer sysctl/proc file value safely.""" - try: - with open(path, "r", encoding="utf-8", errors="ignore") as f: - raw = f.read().strip() - return int(raw or default) - except Exception: - return int(default) - -def read_proc_interrupt_total(): - """Read total interrupts count from /proc/stat.""" - try: - with open("/proc/stat", "r", encoding="utf-8", errors="ignore") as f: - for line in f: - if line.startswith("intr "): - parts = line.strip().split() - if len(parts) >= 2: - return int(parts[1]) - except Exception: - return 0 - return 0 - -def collect_entropy_cloud_status(): - """ - Collect Linux random subsystem entropy status and source activity. - This is a best-effort realtime heuristic for UI visualization. - """ - now = time.time() - entropy_bits = read_sysctl_int("/proc/sys/kernel/random/entropy_avail", 0) - pool_size_bits = read_sysctl_int("/proc/sys/kernel/random/poolsize", 256) - read_threshold = read_sysctl_int("/proc/sys/kernel/random/read_wakeup_threshold", 128) - write_threshold = read_sysctl_int("/proc/sys/kernel/random/write_wakeup_threshold", 64) - - try: - disk = psutil.disk_io_counters() - except Exception: - disk = None - try: - net = psutil.net_io_counters() - except Exception: - net = None - intr_total = read_proc_interrupt_total() - - prev_ts = ENTROPY_PREV.get("timestamp") - dt = max(now - prev_ts, 0.001) if prev_ts else None - - disk_read_now = int(getattr(disk, "read_bytes", 0) or 0) - disk_write_now = int(getattr(disk, "write_bytes", 0) or 0) - net_sent_now = int(getattr(net, "bytes_sent", 0) or 0) - net_recv_now = int(getattr(net, "bytes_recv", 0) or 0) - - if dt: - disk_delta = max( - (disk_read_now - int(ENTROPY_PREV.get("disk_read_bytes") or disk_read_now)) - + (disk_write_now - int(ENTROPY_PREV.get("disk_write_bytes") or disk_write_now)), - 0 - ) - net_delta = max( - (net_sent_now - int(ENTROPY_PREV.get("net_sent_bytes") or net_sent_now)) - + (net_recv_now - int(ENTROPY_PREV.get("net_recv_bytes") or net_recv_now)), - 0 - ) - intr_delta = max(intr_total - int(ENTROPY_PREV.get("interrupt_total") or intr_total), 0) - else: - disk_delta = 0 - net_delta = 0 - intr_delta = 0 - - ENTROPY_PREV["timestamp"] = now - ENTROPY_PREV["disk_read_bytes"] = disk_read_now - ENTROPY_PREV["disk_write_bytes"] = disk_write_now - ENTROPY_PREV["net_sent_bytes"] = net_sent_now - ENTROPY_PREV["net_recv_bytes"] = net_recv_now - ENTROPY_PREV["interrupt_total"] = intr_total - - def scale_intensity(rate_value, scale): - return int(max(0, min(100, (float(rate_value) / float(scale)) * 100.0))) - - disk_rate = (disk_delta / dt) if dt else 0 - net_rate = (net_delta / dt) if dt else 0 - intr_rate = (intr_delta / dt) if dt else 0 - - irq_intensity = scale_intensity(intr_rate, 25000) - disk_intensity = scale_intensity(disk_rate, 80 * 1024 * 1024) - net_intensity = scale_intensity(net_rate, 120 * 1024 * 1024) - hwrng_intensity = 68 if entropy_bits > max(read_threshold, 128) else 34 - - sources = [ - { - "source": "interrupt timing", - "intensity": irq_intensity, - "status": "active" if irq_intensity >= 25 else "low" - }, - { - "source": "disk IO", - "intensity": disk_intensity, - "status": "active" if disk_intensity >= 18 else "low" - }, - { - "source": "network timing", - "intensity": net_intensity, - "status": "active" if net_intensity >= 18 else "low" - }, - { - "source": "hardware RNG", - "intensity": hwrng_intensity, - "status": "active" if hwrng_intensity >= 50 else "limited" - } - ] - - source_avg = int(sum(s["intensity"] for s in sources) / max(len(sources), 1)) - entropy_pct = max(0.0, min(1.0, float(entropy_bits) / max(float(pool_size_bits), 1.0))) - particle_density = max(16, min(84, int(18 + entropy_pct * 42 + source_avg * 0.35))) - key_birth_rate = round(0.6 + entropy_pct * 9.4 + source_avg * 0.06, 2) - - crng_state = "ready" if entropy_bits >= max(read_threshold, 128) else "warming" - random_state = "stable" if entropy_bits >= max(write_threshold, 64) else "refilling" - - return { - "entropy_pool_bits": int(entropy_bits), - "entropy_pool_size_bits": int(pool_size_bits), - "crng_state": crng_state, - "random_subsystem_state": random_state, - "particle_density": int(particle_density), - "key_birth_rate_est": float(key_birth_rate), - "sources": sources, - "read_wakeup_threshold": int(read_threshold), - "write_wakeup_threshold": int(write_threshold), - "mode": "live-heuristic" - } - -def collect_algorithm_requesters(items, kernel_clients): - """Infer likely requestor objects that trigger algorithm competition.""" - algo_map = {"aes": {}, "sha": {}, "chacha20": {}} - client_boost_rules = { - "aes": {"kTLS", "dm-crypt", "AF_ALG", "IPsec/XFRM"}, - "sha": {"kTLS", "AF_ALG", "IPsec/XFRM"}, - "chacha20": {"WireGuard", "AF_ALG"} - } - - for item in items or []: - process_name = str(item.get("process", "unknown")).lower() or "unknown" - protocol = str(item.get("protocol", "")).upper() - algorithm = str(item.get("algorithm", "")).upper() - status = str(item.get("status", "")).upper() - if status == "LISTEN": - continue - - matched_algorithms = set() - if "AES" in algorithm or protocol == "TLS": - matched_algorithms.add("aes") - if "SHA" in algorithm or protocol == "TLS": - matched_algorithms.add("sha") - if "CHACHA" in algorithm or protocol in {"WIREGUARD", "SSH"}: - matched_algorithms.add("chacha20") - if not matched_algorithms and protocol == "CRYPTO API": - matched_algorithms.update(["aes", "sha"]) - - for algo_key in matched_algorithms: - key = f"process:{process_name}" - bucket = algo_map[algo_key].setdefault(key, { - "name": process_name, - "kind": "process", - "score": 0 - }) - bucket["score"] += 1 - - for client in kernel_clients or []: - name = str(client.get("name", "")).strip() - flows = int(client.get("active_flows", 0) or 0) - if not name or flows <= 0: - continue - for algo_key, allowed_clients in client_boost_rules.items(): - if name not in allowed_clients: - continue - key = f"client:{name}" - bucket = algo_map[algo_key].setdefault(key, { - "name": name, - "kind": "kernel-client", - "score": 0 - }) - # Kernel clients are presented as primary requestor objects. - bucket["score"] += max(2, flows) - - result = {} - for algo_key, raw in algo_map.items(): - ranked = sorted(raw.values(), key=lambda x: x.get("score", 0), reverse=True) - if not ranked: - ranked = [{ - "name": "user/kernel request", - "kind": "generic", - "score": 1 - }] - result[algo_key] = ranked[:4] - return result - -def build_crypto_decision_pipelines(algorithm_competitions, kernel_clients, hw_offload, algorithm_requesters): - """Build visual decision pipeline metadata for each algorithm family.""" - hw_active = [h.get("engine") for h in (hw_offload or []) if h.get("status") == "active"] - hw_available = [h.get("engine") for h in (hw_offload or []) if h.get("status") == "available"] - capability_hint = ", ".join(hw_active[:2] or hw_available[:2]) if (hw_active or hw_available) else "generic-cpu-only" - - tfm_lookup_map = { - "AES": "crypto_alloc_skcipher(aes)", - "SHA": "crypto_alloc_shash(sha*)", - "CHACHA20": "crypto_alloc_skcipher(chacha20)" - } - - pipelines = {} - for key, comp in (algorithm_competitions or {}).items(): - request = str(comp.get("request", key)).upper() - impls = comp.get("implementations", []) or [] - shortlist = [str(x.get("name", "unknown")) for x in impls[:3]] - requesters = list((algorithm_requesters or {}).get(key, [])) - top_requester = requesters[0] if requesters else {"name": "user/kernel request", "kind": "generic"} - request_origin = f"{top_requester.get('kind', 'generic')}: {top_requester.get('name', 'unknown')}" - selected_driver = str((comp.get("selected") or {}).get("name", "unknown")) - fallback_driver = next((name for name in shortlist if "generic" in name.lower()), shortlist[-1] if shortlist else "none") - selected_is_generic = "generic" in selected_driver.lower() - selected_source = str((comp.get("selected") or {}).get("source", "kernel")).lower() - fallback_active = selected_is_generic and len(shortlist) > 1 - - pipelines[key] = { - "request": request, - "request_origin": request_origin, - "requesters": requesters, - "tfm_lookup": tfm_lookup_map.get(request, f"crypto_lookup({request.lower()})"), - "impl_shortlist": shortlist, - "priority_check": "max priority wins", - "capability_check": capability_hint, - "selected_driver": selected_driver, - "fallback_driver": fallback_driver, - "fallback_active": bool(fallback_active), - "fallback_reason": "higher-priority impl unavailable or unsupported" if fallback_active else "not-triggered", - "source": selected_source - } - return pipelines - def collect_crypto_realtime(): - return _crypto_security_service.collect_crypto_realtime( - callbacks={ - "infer_crypto_protocol": infer_crypto_protocol, - "is_likely_crypto_actor": is_likely_crypto_actor, - "infer_tls_terminator": infer_tls_terminator, - "collect_algorithm_competition": collect_algorithm_competition, - "parse_proc_crypto_entries": parse_proc_crypto_entries, - "collect_kernel_crypto_clients": collect_kernel_crypto_clients, - "collect_hw_offload_status": collect_hw_offload_status, - "collect_sync_async_queue": collect_sync_async_queue, - "collect_algorithm_requesters": collect_algorithm_requesters, - "build_crypto_decision_pipelines": build_crypto_decision_pipelines, - "collect_entropy_cloud_status": collect_entropy_cloud_status, - }, - crypto_prev=CRYPTO_PREV, - ) + return _crypto_security_service.collect_crypto_realtime(crypto_prev=CRYPTO_PREV, entropy_prev=ENTROPY_PREV) def collect_security_realtime(): return _crypto_security_service.collect_security_realtime(security_prev=SECURITY_PREV)