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/evaluator2misp.py

@@ -80,6 +80,8 @@ def create_object(db_row):
     status = db_row[6]
     classification = db_row[7]
     source = db_row[8]
+    latest_content_hash = db_row[9]
+    content_size = db_row[10]
     ip, domain = extract_ip_domain_port(url)
 
     new_object = MISPObject("url-honeypot-detection", misp_objects_path_custom="/etc/url_evaluator/misp_objects/")
@@ -102,10 +104,18 @@ def create_object(db_row):
     if hash:
         new_object.add_attribute(object_relation="hash", simple_value=hash, Attribute={"type": "sha1", "value": hash})
 
+    # Add SHA-256 / latest content hash if present
+    if latest_content_hash:
+        new_object.add_attribute(object_relation="sha256", simple_value=latest_content_hash, Attribute={"type": "sha256", "value": latest_content_hash})
+
     # Add file type if present
     if file_mime_type:
         new_object.add_attribute(object_relation="mime-type", simple_value=file_mime_type, Attribute={"type": "mime-type", "value": file_mime_type})
 
+    # Add content size if present
+    if content_size is not None:
+        new_object.add_attribute(object_relation="size-in-bytes", simple_value=content_size, Attribute={"type": "size-in-bytes", "value": content_size})
+
     # Add threat label if present
     if threat_label:
         new_object.add_attribute(object_relation="malware-family", simple_value=threat_label, Attribute={"type": "text", "value": threat_label})
@@ -143,11 +153,11 @@ def sync_urls(misp, db):
         u.threat_label,
         u.status,
         u.classification,
-        COALESCE(GROUP_CONCAT(us.source, ', '), 'Unknown')
+        COALESCE((SELECT GROUP_CONCAT(source, ', ') FROM (SELECT DISTINCT source FROM url_source WHERE url = u.url)), 'Unknown'),
+        u.latest_content_hash,
+        u.content_size
     FROM urls u
-    LEFT JOIN url_source us ON us.url = u.url
-    WHERE u.classification IN ('malicious', 'miner') AND u.last_seen >= ?
-    GROUP BY u.url;
+    WHERE u.classification IN ('malicious', 'miner') AND u.last_seen >= ?;
     """, (cutoff_date,)).fetchall()
     if not rows:
         logger.info("No malicious URLs")

bin/evaluator2misp_bysource.py

@@ -80,6 +80,8 @@ def create_object(db_row):
     status = db_row[6]
     classification = db_row[7]
     source = db_row[8]
+    latest_content_hash = db_row[9]
+    content_size = db_row[10]
     ip, domain = extract_ip_domain_port(url)
 
     new_object = MISPObject("url-honeypot-detection", misp_objects_path_custom="/etc/url_evaluator/misp_objects/")
@@ -102,10 +104,18 @@ def create_object(db_row):
     if hash:
         new_object.add_attribute(object_relation="hash", simple_value=hash, Attribute={"type": "sha1", "value": hash})
 
+    # Add SHA-256 / latest content hash if present
+    if latest_content_hash:
+        new_object.add_attribute(object_relation="sha256", simple_value=latest_content_hash, Attribute={"type": "sha256", "value": latest_content_hash})
+
     # Add file type if present
     if file_mime_type:
         new_object.add_attribute(object_relation="mime-type", simple_value=file_mime_type, Attribute={"type": "mime-type", "value": file_mime_type})
 
+    # Add content size if present
+    if content_size is not None:
+        new_object.add_attribute(object_relation="size-in-bytes", simple_value=content_size, Attribute={"type": "size-in-bytes", "value": content_size})
+
     # Add threat label if present
     if threat_label:
         new_object.add_attribute(object_relation="malware-family", simple_value=threat_label, Attribute={"type": "text", "value": threat_label})
@@ -155,11 +165,11 @@ def sync_urls(misp, db):
         u.threat_label,
         u.status,
         u.classification,
-        COALESCE(GROUP_CONCAT(us.source, ', '), 'Unknown')
+        COALESCE((SELECT GROUP_CONCAT(source, ', ') FROM (SELECT DISTINCT source FROM url_source WHERE url = u.url)), 'Unknown'),
+        u.latest_content_hash,
+        u.content_size
     FROM urls u
-    LEFT JOIN url_source us ON us.url = u.url
-    WHERE u.classification IN ('malicious', 'miner') AND u.last_seen >= ?
-    GROUP BY u.url;
+    WHERE u.classification IN ('malicious', 'miner') AND u.last_seen >= ?;
     """, (cutoff_date,)).fetchall()
     if not rows:
         logger.info("No malicious URLs")

bin/evaluator2urlhaus.py

@@ -17,36 +17,54 @@
 def evaluator2urlhaus():
     logger.info("Job started")
 
-    # Load active malicious URLs
+    # Load active malicious URLs with latest content hash / sha256
     with SQLiteWrapper(config.db_path) as db:
-        urls = db.execute("SELECT url FROM urls WHERE classification='malicious' AND status='active'").fetchall()
-    if not urls:
+        rows = db.execute("""
+            SELECT url, hash, latest_content_hash, file_mime_type, content_size
+            FROM urls
+            WHERE classification='malicious' AND status='active'
+        """).fetchall()
+    if not rows:
         logger.info("No URLs to send)")
         return
-    logger.info(f"Found {len(urls)} malicious URLs")
+    logger.info(f"Found {len(rows)} malicious URLs")
 
     # Load URLhaus blacklist
     content = requests.get(config.urlhaus_blacklist_url).content.decode("utf-8")
     blacklist = [line for line in content.splitlines() if not line.startswith("#")]
 
     # Send URLs to URLhaus
     cnt_submissions = 0
-    for url in urls:
-        if url[0] in blacklist:
+    for row in rows:
+        url = row[0]
+        sha1 = row[1]
+        sha256 = row[2]
+        file_mime_type = row[3]
+        content_size = row[4]
+        if url in blacklist:
             continue  # do not send URLs that are already blacklisted
+        submission = {
+            'url': url,
+            'threat': 'malware_download'
+        }
+        if sha256:
+            submission['sha256'] = sha256
+        if sha1:
+            submission['sha1'] = sha1
+        if file_mime_type:
+            submission['content_type'] = file_mime_type
+        if content_size is not None:
+            submission['filesize'] = content_size
         json_data = {
             'token': config.urlhaus_key,
             'anonymous': '0',
-            'submission': [{
-                'url': url[0],
-                'threat': 'malware_download'
-            }]
+            'submission': [submission]
         }
         r = requests.post(config.urlhaus_submit_url, json=json_data, timeout=30, headers={"Content-Type": "application/json"})
         if r.status_code == 200:
             cnt_submissions += 1
 
-    logger.info(f"Sent {cnt_submissions} new submissions (out of {len(urls)} URLs)")
+    logger.info(f"Sent {cnt_submissions} new submissions (out of {len(rows)} URLs)")
     logger.info("Job finished")
 
 def sigint_handler(signum, frame):

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