| |
| |
|
|
| import time |
| import threading |
| from collections import deque |
|
|
| class AnomalousForensicRingBuffer: |
| """ |
| The Organism: Anomalous Forensic Ring Buffer [LK] |
| Provides a fixed-size, RAM-native rolling ring buffer to capture |
| unrecognized zero-day anomalies or suspicious payloads that bypass |
| fast-path Bloom filters, satisfying post-mortem forensic requirements. |
| """ |
| def __init__(self, max_buffer_size: int = 1000): |
| self._lock = threading.Lock() |
| self._max_size = max_buffer_size |
| self._buffer = deque(maxlen=max_buffer_size) |
| self._captured_count = 0 |
|
|
| def capture_anomaly(self, payload: dict, anomaly_score: float) -> None: |
| """Captures unrecognized or suspicious raw payloads into volatile memory.""" |
| with self._lock: |
| entry = { |
| "timestamp": time.time(), |
| "anomaly_score": anomaly_score, |
| "payload": payload, |
| "signature": "[LK]" |
| } |
| self._buffer.append(entry) |
| self._captured_count += 1 |
|
|
| def drain_buffer(self) -> list: |
| """ |
| Drains the ring buffer during idle CPU cycles for the Local AI Observer |
| to structure and lazily write out to encrypted forensic storage. |
| """ |
| with self._lock: |
| items = list(self._buffer) |
| self._buffer.clear() |
| return items |
|
|
| def get_buffer_status(self) -> dict: |
| with self._lock: |
| return { |
| "current_items_in_ram": len(self._buffer), |
| "max_capacity": self._max_size, |
| "total_anomalies_captured": self._captured_count, |
| "signature": "[LK]" |
| } |
|
|
| if __name__ == "__main__": |
| ring = AnomalousForensicRingBuffer(max_buffer_size=5) |
| print("=== THE ORGANISM: FORENSIC RING BUFFER INITIALIZED [LK] ===") |
| |
| |
| ring.capture_anomaly({"src_ip": "192.168.1.99", "raw_bytes": "GET /admin/hidden_vector_0x99"}, anomaly_score=0.98) |
| ring.capture_anomaly({"src_ip": "10.0.0.42", "raw_bytes": "POST /api/v1/overflow_test"}, anomaly_score=0.85) |
| |
| print("Buffer Status:", ring.get_buffer_status()) |
| print("Draining for AI Observer analysis:", ring.drain_buffer()) |
|
|
