Spaces:
Running
Running
| """ | |
| In-memory cache for forensic analysis results. | |
| Why caching? | |
| - Image forensics is CPU-intensive (FFT, hashing, EXIF) | |
| - Same file uploaded twice = wasted computation | |
| - SHA-256 hash = unique file fingerprint | |
| Privacy note: | |
| - Cache stores results only, never file bytes | |
| - Results contain no personal data | |
| - Cache cleared on server restart (no persistence) | |
| """ | |
| import copy | |
| import hashlib | |
| import threading | |
| from typing import Dict, Optional, Any, Union | |
| from datetime import datetime, timedelta | |
| from backend.core.logger import setup_logger | |
| from backend.core.config import settings | |
| logger = setup_logger(__name__) | |
| # Max cached results (prevents memory abuse) | |
| MAX_CACHE_SIZE = settings.MAX_CACHE_SIZE | |
| # Cache TTL: results expire after 1 hour | |
| CACHE_TTL_MINUTES = settings.CACHE_TTL_MINUTES | |
| class ForensicsCache: | |
| """ | |
| Thread-safe in-memory cache for forensic results. | |
| Key: SHA-256 hash of file bytes | |
| Value: forensic report + timestamp | |
| All reads and writes are protected by a single RLock. | |
| TTL expiry is checked inside the lock to prevent TOCTOU races. | |
| """ | |
| def __init__(self): | |
| self._lock = threading.RLock() | |
| self._cache: Dict[str, Dict[str, Any]] = {} | |
| logger.info("Forensics cache initialized") | |
| def _compute_key(self, file_identifier: Union[bytes, str]) -> str: | |
| """ | |
| Compute SHA-256 hash as cache key. | |
| Accepts pre-computed hash (str) to avoid redundant hashing. | |
| """ | |
| if isinstance(file_identifier, str): | |
| return file_identifier | |
| return hashlib.sha256(file_identifier).hexdigest() | |
| def get(self, file_identifier: Union[bytes, str]) -> Optional[Dict[str, Any]]: | |
| """ | |
| Retrieve cached result if available and not expired. | |
| TTL check and deletion are performed inside the lock (TOCTOU-safe). | |
| """ | |
| key = self._compute_key(file_identifier) | |
| with self._lock: | |
| if key not in self._cache: | |
| logger.info(f"Cache MISS: {key[:16]}...") | |
| return None | |
| entry = self._cache[key] | |
| age = datetime.now() - entry.get("cached_at", datetime.now()) | |
| if age > timedelta(minutes=CACHE_TTL_MINUTES): | |
| self._cache.pop(key, None) | |
| logger.info(f"Cache EXPIRED: {key[:16]}...") | |
| return None | |
| logger.info( | |
| f"Cache HIT: {key[:16]}... " | |
| f"(age={age.seconds}s, cache_size={len(self._cache)})" | |
| ) | |
| result = copy.deepcopy(entry["report"]) | |
| result["cache_info"] = { | |
| "cached": True, | |
| "age_seconds": age.seconds, | |
| "cache_hit": True, | |
| } | |
| return result | |
| def set(self, file_identifier: Union[bytes, str], report: Dict[str, Any]) -> None: | |
| """ | |
| Store forensic report in cache. | |
| Evicts oldest entry if cache is full. | |
| All operations are inside the lock (thread-safe). | |
| """ | |
| key = self._compute_key(file_identifier) | |
| with self._lock: | |
| if len(self._cache) >= MAX_CACHE_SIZE: | |
| oldest_key = min( | |
| self._cache, | |
| key=lambda k: self._cache[k]["cached_at"] | |
| ) | |
| del self._cache[oldest_key] | |
| logger.info(f"Cache EVICT: {oldest_key[:16]}...") | |
| self._cache[key] = { | |
| "report": report, | |
| "cached_at": datetime.now(), | |
| } | |
| logger.info( | |
| f"Cache SET: {key[:16]}... " | |
| f"(cache_size={len(self._cache)})" | |
| ) | |
| def size(self) -> int: | |
| """Return current number of cached entries.""" | |
| with self._lock: | |
| return len(self._cache) | |
| def clear(self) -> None: | |
| """Clear all cached entries.""" | |
| with self._lock: | |
| count = len(self._cache) | |
| self._cache.clear() | |
| logger.info(f"Cache CLEARED: {count} entries removed") | |
| # Singleton instance — shared across all requests | |
| forensics_cache = ForensicsCache() | |