new features

bin/evaluator.py

@@ -18,6 +18,7 @@
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')))
 from common.config import Config
 from common.db import SQLiteWrapper
+from common.db_helpers import persist_content_snapshot, update_url_field
 from common.utils import is_valid, extract_commands, process_new_session
 
 
@@ -103,34 +104,42 @@ def search_for_nested_urls(content, src_url):
 
 def analyze_content(url):
     """
-    Download content from given URL and check its hash on VirusTotal / MalwareBazaar
+    Download content from given URL, persist a snapshot, and check its hash on
+    VirusTotal / MalwareBazaar.
     """
 
     try:
         with requests.get(url, stream=True, proxies=proxies, timeout=10) as response:
             if not response.ok:
                 return dict(classification="unreachable", classification_reason=f"Status code {response.status_code}")
-            if (content_size := response.headers.get('Content-Length')) is None:
+            if (content_size_header := response.headers.get('Content-Length')) is None:
                 return dict(classification="unclassified", classification_reason="No content")
-            if (content_size_mb := int(content_size) / (1024 ** 2)) > config.max_file_size:
+            if (content_size_mb := int(content_size_header) / (1024 ** 2)) > config.max_file_size:
                 return dict(classification="unclassified", classification_reason=f"File too large: {content_size_mb:.2f} MB")
 
+            content = response.content
+
             # Determine file type
             file_type = ""
             if "content-type" in response.headers:
                 file_type = response.headers['content-type'].split(";")[0]
             else:
                 try:
-                    file_type = magic.from_buffer(response.content, mime=True)
+                    file_type = magic.from_buffer(content, mime=True)
                 except Exception as e:
                     logger.debug(f"Couldn't determine file type: {e}")
 
+            # Persist the downloaded content snapshot and update url_content / urls
+            snapshot = persist_content_snapshot(
+                db, config.content_storage_path, url, response, content, file_type
+            )
+
             # Search the downloaded content for new URLs
             if file_type in ["application/x-sh", "application/x-shellscript",  "text/plain", "text/x-shellscript", "text/x-sh"]:
-                search_for_nested_urls(response.content, url)
+                search_for_nested_urls(content, url)
 
-            sha1 = hashlib.sha1(response.content).hexdigest()
-            result = dict(hash=sha1, content_size=content_size)
+            sha1 = snapshot["hash"]
+            result = dict(hash=sha1, content_size=snapshot["content_size"])
             if file_type:
                 result.update(file_mime_type=file_type)
 
@@ -318,11 +327,17 @@ def sigint_handler(signum, frame):
                 continue
             logger.info(f"URL {url} was classified as {result['classification']}, reason: {result['classification_reason']}")
 
-            # Update DB record
-            items = list(result.items())
-            set_clause = ", ".join([f"{k} = ?" for k, _ in items])
-            params = tuple(v for _, v in items) + (url,)
-            db.execute(f"UPDATE urls SET {set_clause} WHERE url = ?", params)
+            # Update DB record, recording field-level history for key values
+            tracked_fields = {"classification", "classification_reason", "note", "threat_label", "status", "eval_later"}
+            for field, value in result.items():
+                if field in tracked_fields:
+                    update_url_field(db, url, field, value, changed_by="system")
+                else:
+                    db.execute(f"UPDATE urls SET {field} = ? WHERE url = ?", (value, url))
+
+            # last_edit is set by the helper; set a timestamp for system edits too
+            now = datetime.now(timezone.utc).isoformat()
+            db.execute("UPDATE urls SET last_edit = ? WHERE url = ?", (f"system ({now})", url))
 
             # If the URL was classified as malicious, mark all source URLs that led to it as malicious
             if result["classification"] == "malicious":

bin/honeynetasia2evaluator.py

@@ -13,7 +13,7 @@
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')))
 from common.config import Config
 from common.db import SQLiteWrapper
-from common.utils import is_valid, get_domain
+from common.utils import is_valid, get_domain, record_url_source
 
 
 def honeynetasia2evaluator():
@@ -27,6 +27,7 @@ def honeynetasia2evaluator():
     logger.debug("Data successfully downloaded, processing")
 
     num_inserted = 0
+    observed_at = datetime.utcnow().isoformat()
     with SQLiteWrapper(config.db_path) as db:
         for line in response.content.splitlines():
             url = line.decode().strip().replace("hxxp://", "http://")
@@ -39,7 +40,7 @@ def honeynetasia2evaluator():
                     last_seen = excluded.last_seen,
                     occurrences = urls.occurrences + 1;
             """, (url, current_date, current_date, get_domain(url))).rowcount
-            db.execute("INSERT OR IGNORE INTO url_source (url, source) VALUES (?, ?)", (url, "HoneyNet.Asia"))
+            record_url_source(db, url, "HoneyNet.Asia", observed_at=observed_at)
     logger.info(f"{num_inserted} URLs inserted or updated")
     logger.info("Job finished")
 

bin/tpot2evaluator.py

@@ -32,7 +32,10 @@ def tpot2evaluator():
         with SQLiteWrapper(config.db_path) as db:
             for session in response.json():
                 logger.debug(f"Looking for new URLs in '{session['input']}'")
-                if new_urls := process_new_session(db, config, session["input"], None, session["timestamp"], "GEANT T-Pot", None):
+                source_detail = session.get("sensor", session.get("host", None))
+                if new_urls := process_new_session(
+                    db, config, session["input"], None, session["timestamp"], "GEANT T-Pot", source_detail
+                ):
                     logger.info(f"Discovered {len(new_urls)} new URLs: {new_urls}")
     except Exception as e:
         logger.error(f"Error while processing sessions: {type(e).__name__}: {e}")

bin/warden2evaluator.py

@@ -78,7 +78,9 @@ def receiver():
                         logger.debug(f"Looking for new URLs in '{content}'")
 
                         # Search for new URLs
-                        if new_urls := process_new_session(db, config, content, event.get("ID"), event.get("DetectTime"), source, None):
+                        if new_urls := process_new_session(
+                            db, config, content, event.get("ID"), event.get("DetectTime"), source, None
+                        ):
                             logger.info(f"Discovered {len(new_urls)} new URLs (event ID {event.get('ID')}): {new_urls}")
 
 

common/content_storage.py

@@ -0,0 +1,104 @@
+import hashlib
+import os
+from pathlib import Path
+import logging
+
+logger = logging.getLogger("content_storage")
+
+def storage_path(base_dir: str, content_hash: str) -> Path:
+    """
+    Derive storage path from SHA-256 hash: first two hex chars / next two hex chars / rest of hash + .blob
+    Example: abcdef1234... -> content/ab/cd/abcdef1234...blob
+    """
+    if not content_hash or len(content_hash) < 4:
+        raise ValueError("Invalid content hash")
+
+    content_hash = content_hash.lower()
+
+    # SHA-256 hashes are 64 hex characters; validate the whole string.
+    if len(content_hash) != 64 or not all(c in "0123456789abcdef" for c in content_hash):
+        raise ValueError("Invalid SHA-256 content hash")
+
+    dir1 = content_hash[0:2]
+    dir2 = content_hash[2:4]
+    filename = f"{content_hash}.blob"
+
+    return Path(base_dir) / dir1 / dir2 / filename
+
+def save_content(base_dir: str, content: bytes) -> tuple[str, str, str, bool]:
+    """
+    Save content to disk using SHA-256 deduplication.
+    Returns: (sha256, sha1, relative_path, is_new)
+    """
+    sha256 = hashlib.sha256(content).hexdigest()
+    sha1 = hashlib.sha1(content).hexdigest()
+
+    path = storage_path(base_dir, sha256)
+
+    is_new = False
+    if not path.exists():
+        try:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with open(path, "wb") as f:
+                f.write(content)
+            is_new = True
+            logger.debug(f"Saved new content snapshot: {sha256} to {path}")
+        except Exception as e:
+            logger.exception(f"Error saving content to {path}: {e}")
+            raise
+    else:
+        logger.debug(f"Content snapshot {sha256} already exists")
+
+    return sha256, sha1, str(path.relative_to(base_dir)), is_new
+
+def load_content(base_dir: str, content_hash: str) -> bytes:
+    """
+    Load content from disk.
+    """
+    path = storage_path(base_dir, content_hash)
+    if not path.exists():
+        raise FileNotFoundError(f"Content snapshot {content_hash} not found at {path}")
+
+    with open(path, "rb") as f:
+        return f.read()
+
+def content_exists(base_dir: str, content_hash: str) -> bool:
+    """
+    Check if content exists on disk.
+    """
+    return storage_path(base_dir, content_hash).exists()
+
+
+def delete_content(base_dir: str, content_hash: str) -> bool:
+    """
+    Delete a stored content snapshot from disk.
+
+    Returns True if the file existed and was removed, False otherwise.
+    Parent directories created by the storage layout are removed if empty.
+    """
+    path = storage_path(base_dir, content_hash)
+    if not path.exists():
+        return False
+
+    try:
+        path.unlink()
+        logger.debug(f"Deleted content snapshot: {content_hash} from {path}")
+    except Exception as e:
+        logger.exception(f"Error deleting content {content_hash}: {e}")
+        raise
+
+    # Clean up empty parent directories (up to base_dir).
+    try:
+        for parent in path.parents:
+            if parent == Path(base_dir).resolve():
+                break
+            try:
+                parent.rmdir()
+            except OSError:
+                # Directory not empty or already removed; stop ascending.
+                break
+    except Exception:
+        # Non-fatal cleanup; the snapshot itself is already gone.
+        pass
+
+    return True