From 44448f7b05ac03af7d0bc8f1d46cd35510847a9a Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Thu, 23 Jul 2026 15:18:57 +0200 Subject: [PATCH 1/8] feat: new db schema --- install/create_db.sql | 112 +++++++++++++++++++++++++++++++----------- 1 file changed, 84 insertions(+), 28 deletions(-) diff --git a/install/create_db.sql b/install/create_db.sql index 3df10a2..fb83b8e 100644 --- a/install/create_db.sql +++ b/install/create_db.sql @@ -5,33 +5,6 @@ CREATE TABLE sessions idea_id TEXT ); -CREATE TABLE url_session -( - id INTEGER PRIMARY KEY AUTOINCREMENT, - url TEXT REFERENCES urls(url), - session TEXT REFERENCES sessions(session_hash), - - CONSTRAINT url_session_unique UNIQUE (url, session) -); - -CREATE TABLE url_source -( - id INTEGER PRIMARY KEY AUTOINCREMENT, - url TEXT REFERENCES urls(url), - source TEXT, - - CONSTRAINT url_source_unique UNIQUE (url, source) -); - -CREATE TABLE discovered_urls -( - id INTEGER PRIMARY KEY AUTOINCREMENT, - url TEXT REFERENCES urls(url), - src_url TEXT REFERENCES urls(url), - - CONSTRAINT discovered_urls_unique UNIQUE (url, src_url) -); - CREATE TABLE urls ( url TEXT PRIMARY KEY, @@ -53,5 +26,88 @@ CREATE TABLE urls status_changed TEXT DEFAULT 'no' CHECK (status_changed IN ('yes', 'no')), last_edit TEXT, eval_later TEXT DEFAULT 'no' CHECK (eval_later IN ('yes', 'no')), - domain TEXT + domain TEXT, + latest_content_hash TEXT +); + +CREATE TABLE url_source +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + source TEXT, + source_detail TEXT, + origin_url TEXT, + observed_at DATETIME, + idea_id TEXT, + session_hash TEXT REFERENCES sessions(session_hash) ); + +CREATE TABLE content_snapshot +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content_hash TEXT UNIQUE, + url TEXT, + downloaded_at DATETIME, + source_ip TEXT, + server_ip TEXT, + http_status INTEGER, + http_headers TEXT, + mime_type TEXT, + content_size INTEGER, + storage_path TEXT, + sha1 TEXT, + sha256 TEXT +); + +CREATE TABLE url_content +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + content_hash TEXT REFERENCES content_snapshot(content_hash), + first_seen DATETIME, + last_seen DATETIME, + is_latest TEXT CHECK (is_latest IN ('yes', 'no')), + CONSTRAINT url_content_unique UNIQUE (url, content_hash) +); + +CREATE TABLE discovered_urls +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + src_url TEXT REFERENCES urls(url), + discovered_at DATETIME, + CONSTRAINT discovered_urls_unique UNIQUE (url, src_url) +); + +CREATE TABLE url_history +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + changed_at DATETIME, + field TEXT, + old_value TEXT, + new_value TEXT, + changed_by TEXT +); + +CREATE TABLE sandbox_job +( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content_hash TEXT REFERENCES content_snapshot(content_hash), + url TEXT REFERENCES urls(url), + provider TEXT, + external_id TEXT, + status TEXT CHECK (status IN ('pending', 'running', 'completed', 'failed')), + submitted_at DATETIME, + completed_at DATETIME, + report_url TEXT, + report_json TEXT, + requested_by TEXT +); + +CREATE INDEX idx_url_source_lookup ON url_source(url, source, observed_at); +CREATE INDEX idx_url_source_origin ON url_source(origin_url); +CREATE INDEX idx_url_content_latest ON url_content(url, is_latest); +CREATE INDEX idx_url_content_hash ON url_content(content_hash); +CREATE INDEX idx_url_history_url ON url_history(url, changed_at); +CREATE INDEX idx_sandbox_job_status ON sandbox_job(content_hash, status); From 686a219757b91c7ca480b0807f53e5b701249787 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Thu, 23 Jul 2026 18:12:56 +0200 Subject: [PATCH 2/8] Stage 1: database schema migration, config and content storage foundation - Add idempotent install/migrate_db.py for Stage 1 schema - Add common/content_storage.py for SHA-256 based content deduplication - Extend etc/config.yaml with content_storage_path, max_storage_size_gb, sandbox_providers - Create /data/url_evaluator/content in prepare_environment.sh - Run migrate_db.py after create_db.sql in configure_sqlite.sh --- common/content_storage.py | 63 ++++++ etc/config.yaml | 10 + install/configure_sqlite.sh | 3 + install/migrate_db.py | 337 +++++++++++++++++++++++++++++++++ install/prepare_environment.sh | 5 + 5 files changed, 418 insertions(+) create mode 100644 common/content_storage.py create mode 100644 install/migrate_db.py diff --git a/common/content_storage.py b/common/content_storage.py new file mode 100644 index 0000000..ce13b1d --- /dev/null +++ b/common/content_storage.py @@ -0,0 +1,63 @@ +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") + + 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() diff --git a/etc/config.yaml b/etc/config.yaml index b9490a8..7a423a3 100644 --- a/etc/config.yaml +++ b/etc/config.yaml @@ -37,6 +37,16 @@ max_age_invalid: 7 # days # Max size of downloaded content max_file_size: 100 # MB +# Base directory for downloaded content snapshots +content_storage_path: "/data/url_evaluator/content" + +# Optional soft limit / warning threshold for total storage (GB) +# Set to 0 to disable the warning. +max_storage_size_gb: 0 + +# Placeholder list of sandbox providers for future integration +sandbox_providers: [] + # How often should evaluation blacklist be updated bl_update_time: 15 # minutes diff --git a/install/configure_sqlite.sh b/install/configure_sqlite.sh index 4b3821f..538c745 100644 --- a/install/configure_sqlite.sh +++ b/install/configure_sqlite.sh @@ -9,3 +9,6 @@ echob "** Setting up evaluator DB **" sudo -u url_evaluator sqlite3 /data/url_evaluator/db.sqlite < $BASEDIR/create_db.sql chmod 664 /data/url_evaluator/db.sqlite + +echob "** Running DB migration **" +sudo -u url_evaluator python3 $BASEDIR/migrate_db.py -c /etc/url_evaluator/config.yaml diff --git a/install/migrate_db.py b/install/migrate_db.py new file mode 100644 index 0000000..70a14fa --- /dev/null +++ b/install/migrate_db.py @@ -0,0 +1,337 @@ +#!/usr/bin/env python3 +""" +Database migration script for URL Evaluator Stage 1. + +Prepares an existing database for the new data model described in +plans/implementation_plan.md and plans/new_data_model.md: + + - Extends url_source with richer source metadata. + - Adds latest_content_hash to urls. + - Creates content_snapshot, url_content, url_history and sandbox_job tables. + - Adds discovered_at to discovered_urls. + - Creates required indexes. + - Migrates existing url_source rows and creates placeholder content_snapshot + records for URLs that already have a hash / file_mime_type. + +The script is idempotent and can be run multiple times safely. +""" + +import argparse +import hashlib +import logging +import os +import sys +from datetime import datetime, timezone + +# Add the project root to the import path so common.* can be found. +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 + + +LOGFORMAT = "%(asctime)-15s %(name)s [%(levelname)s] %(message)s" +LOGDATEFORMAT = "%Y-%m-%dT%H:%M:%S" +logging.basicConfig(level=logging.INFO, format=LOGFORMAT, datefmt=LOGDATEFORMAT) +logger = logging.getLogger("migrate_db") + + +def _table_exists(cursor, table: str) -> bool: + cursor.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)) + return cursor.fetchone() is not None + + +def _column_exists(cursor, table: str, column: str) -> bool: + cursor.execute(f"PRAGMA table_info({table})") + return any(row[1] == column for row in cursor.fetchall()) + + +def _index_exists(cursor, index: str) -> bool: + cursor.execute("SELECT 1 FROM sqlite_master WHERE type='index' AND name=?", (index,)) + return cursor.fetchone() is not None + + +def _add_column(cursor, table: str, column: str, definition: str) -> None: + if not _column_exists(cursor, table, column): + logger.info(f"Adding column {table}.{column}") + cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}") + + +def ensure_schema(db) -> None: + """Create missing tables / columns / indexes (idempotent).""" + cursor = db.cursor + + # ------------------------------------------------------------------ + # Core tables (kept from the original schema) + # ------------------------------------------------------------------ + cursor.execute(""" + CREATE TABLE IF NOT EXISTS sessions ( + session_hash TEXT PRIMARY KEY, + session TEXT, + idea_id TEXT + ); + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS urls ( + url TEXT PRIMARY KEY, + first_seen DATE, + last_seen DATE, + hash TEXT, + classification TEXT DEFAULT 'unclassified' + CHECK (classification IN ('malicious', 'harmless', 'unreachable', 'unclassified', 'invalid', 'miner')), + classification_reason TEXT DEFAULT 'Waiting for evaluation', + note TEXT, + reported TEXT DEFAULT 'no' CHECK (reported IN ('yes', 'no')), + occurrences INTEGER DEFAULT 1, + vt_stats TEXT, + evaluated TEXT DEFAULT 'no' CHECK (evaluated IN ('yes', 'no')), + file_mime_type TEXT, + content_size INTEGER, + threat_label TEXT, + status TEXT DEFAULT 'unknown' + CHECK (status IN ('active', 'inactive', 'unknown')), + last_active DATE, + status_changed TEXT DEFAULT 'no' CHECK (status_changed IN ('yes', 'no')), + last_edit TEXT, + eval_later TEXT DEFAULT 'no' CHECK (eval_later IN ('yes', 'no')), + domain TEXT + ); + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS url_source ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + source TEXT, + source_detail TEXT, + origin_url TEXT, + observed_at DATETIME, + idea_id TEXT, + session_hash TEXT REFERENCES sessions(session_hash) + ); + """) + + # The code base references discovered_urls, so keep that name. + cursor.execute(""" + CREATE TABLE IF NOT EXISTS discovered_urls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + src_url TEXT REFERENCES urls(url), + discovered_at DATETIME, + CONSTRAINT discovered_urls_unique UNIQUE (url, src_url) + ); + """) + + # ------------------------------------------------------------------ + # New tables for Stage 1 + # ------------------------------------------------------------------ + cursor.execute(""" + CREATE TABLE IF NOT EXISTS content_snapshot ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content_hash TEXT UNIQUE, + url TEXT, + downloaded_at DATETIME, + source_ip TEXT, + server_ip TEXT, + http_status INTEGER, + http_headers TEXT, + mime_type TEXT, + content_size INTEGER, + storage_path TEXT, + sha1 TEXT, + sha256 TEXT + ); + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS url_content ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + content_hash TEXT REFERENCES content_snapshot(content_hash), + first_seen DATETIME, + last_seen DATETIME, + is_latest TEXT CHECK (is_latest IN ('yes', 'no')), + CONSTRAINT url_content_unique UNIQUE (url, content_hash) + ); + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS url_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT REFERENCES urls(url), + changed_at DATETIME, + field TEXT, + old_value TEXT, + new_value TEXT, + changed_by TEXT + ); + """) + + cursor.execute(""" + CREATE TABLE IF NOT EXISTS sandbox_job ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + content_hash TEXT REFERENCES content_snapshot(content_hash), + url TEXT REFERENCES urls(url), + provider TEXT, + external_id TEXT, + status TEXT CHECK (status IN ('pending', 'running', 'completed', 'failed')), + submitted_at DATETIME, + completed_at DATETIME, + report_url TEXT, + report_json TEXT, + requested_by TEXT + ); + """) + + # ------------------------------------------------------------------ + # Add new columns to existing tables when migrating an old DB + # ------------------------------------------------------------------ + _add_column(cursor, "urls", "latest_content_hash", "TEXT") + _add_column(cursor, "url_source", "source_detail", "TEXT") + _add_column(cursor, "url_source", "origin_url", "TEXT") + _add_column(cursor, "url_source", "observed_at", "DATETIME") + _add_column(cursor, "url_source", "idea_id", "TEXT") + _add_column(cursor, "url_source", "session_hash", "TEXT") + _add_column(cursor, "discovered_urls", "discovered_at", "DATETIME") + + # ------------------------------------------------------------------ + # Indexes + # ------------------------------------------------------------------ + indexes = [ + ("idx_url_source_lookup", "CREATE INDEX idx_url_source_lookup ON url_source(url, source, observed_at)"), + ("idx_url_source_origin", "CREATE INDEX idx_url_source_origin ON url_source(origin_url)"), + ("idx_url_content_latest", "CREATE INDEX idx_url_content_latest ON url_content(url, is_latest)"), + ("idx_url_content_hash", "CREATE INDEX idx_url_content_hash ON url_content(content_hash)"), + ("idx_url_history_url", "CREATE INDEX idx_url_history_url ON url_history(url, changed_at)"), + ("idx_sandbox_job_status", "CREATE INDEX idx_sandbox_job_status ON sandbox_job(content_hash, status)"), + ] + for name, sql in indexes: + if not _index_exists(cursor, name): + logger.info(f"Creating index {name}") + cursor.execute(sql) + + db.conn.commit() + + +def migrate_url_source(db) -> None: + """Backfill url_source metadata from urls / discovered_urls.""" + cursor = db.cursor + + # For every url_source row that has not been observed_at yet, set it from + # the URL's first_seen (or now as a last resort). + cursor.execute(""" + SELECT s.id, s.url, u.first_seen + FROM url_source s + JOIN urls u ON u.url = s.url + WHERE s.observed_at IS NULL; + """) + rows = cursor.fetchall() + if rows: + logger.info(f"Backfilling observed_at for {len(rows)} url_source rows") + now = datetime.now(timezone.utc).isoformat() + for row_id, url, first_seen in rows: + observed_at = first_seen or now + cursor.execute( + "UPDATE url_source SET observed_at = ? WHERE id = ?", + (observed_at, row_id) + ) + + # Backfill origin_url from discovered_urls if url_source has no origin_url + # but discovered_urls has a matching src_url and url. + cursor.execute(""" + SELECT DISTINCT d.url, d.src_url + FROM discovered_urls d + WHERE d.src_url IS NOT NULL + AND EXISTS ( + SELECT 1 FROM url_source s + WHERE s.url = d.url AND s.origin_url IS NULL + ); + """) + discovered = cursor.fetchall() + if discovered: + logger.info(f"Backfilling origin_url for {len(discovered)} discovered URLs") + for url, src_url in discovered: + cursor.execute( + "UPDATE url_source SET origin_url = ? WHERE url = ? AND origin_url IS NULL", + (src_url, url) + ) + + db.conn.commit() + + +def migrate_content_snapshots(db) -> None: + """ + Create placeholder content_snapshot and url_content rows for existing URLs + that already have a hash / file_mime_type but no snapshot yet. + """ + cursor = db.cursor + + cursor.execute(""" + SELECT u.url, u.hash, u.file_mime_type, u.content_size, u.first_seen + FROM urls u + WHERE u.hash IS NOT NULL + AND u.latest_content_hash IS NULL; + """) + rows = cursor.fetchall() + if not rows: + return + + logger.info(f"Creating placeholder snapshots for {len(rows)} URLs with existing hash") + now = datetime.now(timezone.utc).isoformat() + + for url, sha1, mime_type, content_size, first_seen in rows: + # We do not have the actual content, so we cannot compute SHA-256. + # Use a deterministic synthetic hash so the same SHA-1 maps to the same + # placeholder snapshot. This preserves backward compatibility without + # fabricating real SHA-256 hashes. + synthetic = hashlib.sha256(f"placeholder:{sha1}".encode()).hexdigest() + + cursor.execute( + "INSERT OR IGNORE INTO content_snapshot (content_hash, url, downloaded_at, mime_type, content_size, storage_path, sha1, sha256) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (synthetic, url, first_seen or now, mime_type, content_size, None, sha1, synthetic) + ) + + # Only insert url_content if not already present for this URL/hash pair. + cursor.execute( + "SELECT 1 FROM url_content WHERE url = ? AND content_hash = ?", + (url, synthetic) + ) + if cursor.fetchone() is None: + cursor.execute( + """ + INSERT INTO url_content (url, content_hash, first_seen, last_seen, is_latest) + VALUES (?, ?, ?, ?, 'yes') + """, + (url, synthetic, first_seen or now, first_seen or now) + ) + + cursor.execute( + "UPDATE urls SET latest_content_hash = ? WHERE url = ?", + (synthetic, url) + ) + + db.conn.commit() + + +def main(): + parser = argparse.ArgumentParser(description="Migrate URL Evaluator database to Stage 1 schema") + parser.add_argument("--config", "-c", default="/etc/url_evaluator/config.yaml", help="Path to config file") + parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") + args = parser.parse_args() + + if args.verbose: + logger.setLevel(logging.DEBUG) + + config = Config(args.config) + + with SQLiteWrapper(config.db_path) as db: + logger.info(f"Migrating database {config.db_path}") + ensure_schema(db) + migrate_url_source(db) + migrate_content_snapshots(db) + logger.info("Migration finished successfully") + + +if __name__ == "__main__": + main() diff --git a/install/prepare_environment.sh b/install/prepare_environment.sh index 14cceb2..b5850a0 100644 --- a/install/prepare_environment.sh +++ b/install/prepare_environment.sh @@ -33,3 +33,8 @@ chmod -R 775 /var/log/url_evaluator mkdir -p /data/url_evaluator chown -R url_evaluator:url_evaluator /data/url_evaluator/ chmod -R 775 /data/url_evaluator + +# Content storage directory +mkdir -p /data/url_evaluator/content +chown -R url_evaluator:url_evaluator /data/url_evaluator/content/ +chmod -R 775 /data/url_evaluator/content From bb1f0acc11b338237d8342273f9f2e6a68d9ec4c Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Thu, 23 Jul 2026 18:54:18 +0200 Subject: [PATCH 3/8] Add content storage module - Implement SHA-256 based storage path layout. - Add save_content, load_content, content_exists helpers. - Add delete_content helper with empty parent directory cleanup. - Normalise and validate SHA-256 hashes in storage_path. --- common/content_storage.py | 45 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/common/content_storage.py b/common/content_storage.py index ce13b1d..aba12e5 100644 --- a/common/content_storage.py +++ b/common/content_storage.py @@ -12,11 +12,17 @@ def storage_path(base_dir: str, content_hash: str) -> Path: """ 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]: @@ -61,3 +67,38 @@ 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 From 6876ad617707e027824b3c89340000eeb0fa0705 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Fri, 24 Jul 2026 10:56:33 +0200 Subject: [PATCH 4/8] evaluator: persist downloaded content snapshots - Add common/db_helpers.py with persist_content_snapshot(), set_url_latest_content() and record_url_history(). - Extend bin/evaluator.analyze_content() to save downloaded content, metadata and update content_snapshot/url_content/urls/url_history. - Keep existing VT/MB classification logic unchanged. --- bin/evaluator.py | 23 ++++-- common/db_helpers.py | 169 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 7 deletions(-) create mode 100644 common/db_helpers.py diff --git a/bin/evaluator.py b/bin/evaluator.py index f062167..05e062b 100644 --- a/bin/evaluator.py +++ b/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 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) diff --git a/common/db_helpers.py b/common/db_helpers.py new file mode 100644 index 0000000..c29d6f0 --- /dev/null +++ b/common/db_helpers.py @@ -0,0 +1,169 @@ +import json +import logging +from datetime import datetime, timezone + +from common.content_storage import save_content + +LOGFORMAT = "%(asctime)-15s %(name)s [%(levelname)s] %(message)s" +LOGDATEFORMAT = "%Y-%m-%dT%H:%M:%S" +logging.basicConfig(level=logging.INFO, format=LOGFORMAT, datefmt=LOGDATEFORMAT) +logger = logging.getLogger("db_helpers") + + +def record_url_history(db, url, field, old_value, new_value, changed_by="system"): + """ + Record a change to a URL field in url_history. + + If old_value == new_value, no row is inserted. + """ + if old_value == new_value: + return + + now = datetime.now(timezone.utc).isoformat() + db.execute( + """ + INSERT INTO url_history (url, changed_at, field, old_value, new_value, changed_by) + VALUES (?, ?, ?, ?, ?, ?) + """, + (url, now, field, old_value, new_value, changed_by), + ) + + +def set_url_latest_content(db, url, content_hash): + """ + Mark content_hash as the latest snapshot for url. + + All other url_content rows for this URL are marked is_latest = 'no'. + The row for content_hash is inserted (if missing) with first_seen = now + and updated with last_seen = now and is_latest = 'yes'. + """ + now = datetime.now(timezone.utc).isoformat() + + db.execute( + "UPDATE url_content SET is_latest = 'no' WHERE url = ? AND is_latest = 'yes'", + (url,), + ) + db.execute( + """ + INSERT INTO url_content (url, content_hash, first_seen, last_seen, is_latest) + VALUES (?, ?, ?, ?, 'yes') + ON CONFLICT(url, content_hash) DO UPDATE SET + last_seen = excluded.last_seen, + is_latest = 'yes'; + """, + (url, content_hash, now, now), + ) + + +def _extract_socket(response): + """Best-effort extraction of the underlying socket from a requests Response.""" + try: + conn = response.raw._connection + if conn is None: + return None + sock = conn.sock + if sock is None: + return None + return sock + except Exception: + pass + try: + # Some urllib3 versions keep the socket on the HTTPResponse itself. + fp = response.raw._fp + if fp is None: + return None + inner_fp = fp.fp + if inner_fp is None: + return None + return inner_fp._sock + except Exception: + pass + return None + + +def persist_content_snapshot(db, base_dir, url, response, content, mime_type): + """ + Persist a downloaded content snapshot and update url_content / urls. + + Saves the content to disk, inserts a content_snapshot row (if new), + upserts the corresponding url_content row as latest, updates urls with + hash / latest_content_hash / file_mime_type / content_size, and records + a url_history row when the latest_content_hash changes. + + Returns a dict with snapshot metadata: hash (SHA-1), latest_content_hash + (SHA-256), file_mime_type and content_size. + """ + sha256, sha1, storage_path_rel, is_new = save_content(base_dir, content) + downloaded_at = datetime.now(timezone.utc).isoformat() + + source_ip = None + server_ip = None + try: + sock = _extract_socket(response) + if sock is not None: + peer = sock.getpeername() + local = sock.getsockname() + if peer and isinstance(peer[0], str): + server_ip = peer[0] + if local and isinstance(local[0], str): + source_ip = local[0] + except Exception: + pass + + http_status = response.status_code + http_headers = json.dumps(dict(response.headers)) + content_size = len(content) + + if is_new: + db.execute( + """ + INSERT OR IGNORE INTO content_snapshot + (content_hash, url, downloaded_at, source_ip, server_ip, http_status, + http_headers, mime_type, content_size, storage_path, sha1, sha256) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + sha256, + url, + downloaded_at, + source_ip, + server_ip, + http_status, + http_headers, + mime_type, + content_size, + storage_path_rel, + sha1, + sha256, + ), + ) + logger.debug(f"Recorded content snapshot {sha256} for {url}") + + set_url_latest_content(db, url, sha256) + + old_row = db.execute( + "SELECT latest_content_hash FROM urls WHERE url = ?", (url,) + ).fetchone() + old_hash = old_row[0] if old_row else None + + db.execute( + """ + UPDATE urls + SET hash = ?, latest_content_hash = ?, file_mime_type = ?, content_size = ? + WHERE url = ? + """, + (sha1, sha256, mime_type, content_size, url), + ) + + if old_hash != sha256: + record_url_history( + db, url, "latest_content_hash", old_hash, sha256, changed_by="system" + ) + logger.info(f"URL {url} latest content hash changed: {old_hash} -> {sha256}") + + return { + "hash": sha1, + "latest_content_hash": sha256, + "file_mime_type": mime_type, + "content_size": content_size, + } From 952ae34db15c5cc7f848df150571d238383a0703 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Fri, 24 Jul 2026 11:34:38 +0200 Subject: [PATCH 5/8] New feat integration: richer source metadata in ingestion - Add record_url_source() and record_discovered_url() helpers to common/utils.py - Update process_new_session() to populate source_detail, origin_url, observed_at, idea_id and session_hash in url_source and discovered_at in discovered_urls - Update bin/honeynetasia2evaluator.py to record observed_at via record_url_source() - Update bin/tpot2evaluator.py to pass sensor/host as source_detail - Update bin/warden2evaluator.py to pass event metadata through process_new_session() - Ensure migrate_db.py creates unique index on url_source(url, source) for upserts --- bin/honeynetasia2evaluator.py | 5 +-- bin/tpot2evaluator.py | 5 ++- bin/warden2evaluator.py | 4 ++- common/utils.py | 58 +++++++++++++++++++++++++++++++++-- install/migrate_db.py | 7 +++++ 5 files changed, 73 insertions(+), 6 deletions(-) diff --git a/bin/honeynetasia2evaluator.py b/bin/honeynetasia2evaluator.py index aa829e8..efcbf43 100644 --- a/bin/honeynetasia2evaluator.py +++ b/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") diff --git a/bin/tpot2evaluator.py b/bin/tpot2evaluator.py index 4284606..8c72fc0 100644 --- a/bin/tpot2evaluator.py +++ b/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}") diff --git a/bin/warden2evaluator.py b/bin/warden2evaluator.py index b7aeb27..89c7f00 100644 --- a/bin/warden2evaluator.py +++ b/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}") diff --git a/common/utils.py b/common/utils.py index c938155..8ab52a3 100644 --- a/common/utils.py +++ b/common/utils.py @@ -72,6 +72,49 @@ def get_domain(url: str): return None +def record_url_source(db, url, source, source_detail=None, origin_url=None, + observed_at=None, idea_id=None, session_hash=None): + """ + Record an observation of `url` in `url_source`, preserving first-seen time + and updating the last-seen time on duplicate observations. + """ + if observed_at is None: + from datetime import datetime, timezone + observed_at = datetime.now(timezone.utc).isoformat() + + db.execute( + """ + INSERT INTO url_source (url, source, source_detail, origin_url, + observed_at, idea_id, session_hash) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(url, source) DO UPDATE SET + source_detail = COALESCE(excluded.source_detail, url_source.source_detail), + origin_url = COALESCE(excluded.origin_url, url_source.origin_url), + observed_at = COALESCE(url_source.observed_at, excluded.observed_at), + idea_id = COALESCE(excluded.idea_id, url_source.idea_id), + session_hash = COALESCE(excluded.session_hash, url_source.session_hash); + """, + (url, source, source_detail, origin_url, observed_at, idea_id, session_hash), + ) + + +def record_discovered_url(db, url, src_url, discovered_at=None): + """ + Record that `url` was discovered inside the content of `src_url`. + """ + if discovered_at is None: + from datetime import datetime, timezone + discovered_at = datetime.now(timezone.utc).isoformat() + + db.execute( + """ + INSERT OR IGNORE INTO discovered_urls (url, src_url, discovered_at) + VALUES (?, ?, ?) + """, + (url, src_url, discovered_at), + ) + + def process_new_session(db, config, session, idea_id, detect_time, source, source_url): """ Process a new session: @@ -84,7 +127,11 @@ def process_new_session(db, config, session, idea_id, detect_time, source, sourc inserted_urls = [] session_hash = hashlib.md5(session.encode()).hexdigest() + if detect_time is None: + from datetime import datetime, timezone + detect_time = datetime.now(timezone.utc).isoformat() date = detect_time.split("T")[0] + observed_at = detect_time # Extract URLs from shell commands if not (extracted_urls := extract_urls(session)): @@ -100,9 +147,16 @@ def process_new_session(db, config, session, idea_id, detect_time, source, sourc ) for url, occurrences in Counter(extracted_urls).items(): db.execute("INSERT OR IGNORE INTO url_session (url, session) VALUES (?, ?)", (url, session_hash)) - db.execute("INSERT OR IGNORE INTO url_source (url, source) VALUES (?, ?)", (url, source)) + record_url_source( + db, url, source, + source_detail=source_url, + origin_url=source_url, + observed_at=observed_at, + idea_id=idea_id, + session_hash=session_hash, + ) if source_url: - db.execute("INSERT OR IGNORE INTO discovered_urls (url, src_url) VALUES (?, ?)", (url, source_url)) + record_discovered_url(db, url, source_url, discovered_at=observed_at) db.execute( """ INSERT INTO urls (url, first_seen, last_seen, domain) VALUES (?, ?, ?, ?) diff --git a/install/migrate_db.py b/install/migrate_db.py index 70a14fa..b9e3cc8 100644 --- a/install/migrate_db.py +++ b/install/migrate_db.py @@ -187,6 +187,13 @@ def ensure_schema(db) -> None: # ------------------------------------------------------------------ # Add new columns to existing tables when migrating an old DB # ------------------------------------------------------------------ + # Ensure url_source has a unique constraint on (url, source) so that + # ON CONFLICT(url, source) works in record_url_source(). + cursor.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS idx_url_source_unique + ON url_source(url, source); + """) + _add_column(cursor, "urls", "latest_content_hash", "TEXT") _add_column(cursor, "url_source", "source_detail", "TEXT") _add_column(cursor, "url_source", "origin_url", "TEXT") From c0179f79c9071b211920697de6ec77c1c417fcc9 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Fri, 24 Jul 2026 12:01:06 +0200 Subject: [PATCH 6/8] Add URL field history helper and wire it into evaluator and web edits --- bin/evaluator.py | 18 ++++++++++++------ common/db_helpers.py | 25 +++++++++++++++++++++++++ web/main.py | 26 +++++++++++++++++--------- 3 files changed, 54 insertions(+), 15 deletions(-) diff --git a/bin/evaluator.py b/bin/evaluator.py index 05e062b..3060c39 100644 --- a/bin/evaluator.py +++ b/bin/evaluator.py @@ -18,7 +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 +from common.db_helpers import persist_content_snapshot, update_url_field from common.utils import is_valid, extract_commands, process_new_session @@ -327,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": diff --git a/common/db_helpers.py b/common/db_helpers.py index c29d6f0..7533f7c 100644 --- a/common/db_helpers.py +++ b/common/db_helpers.py @@ -29,6 +29,31 @@ def record_url_history(db, url, field, old_value, new_value, changed_by="system" ) +def update_url_field(db, url, field, new_value, changed_by="system"): + """ + Update a single field in urls if the value changed, recording the change + in url_history via record_url_history. + + The caller is responsible for committing. + Returns True if the value was updated, False if unchanged. + """ + # Defensive: ensure the field exists in urls to prevent accidental SQL injection + info = db.execute("PRAGMA table_info(urls)").fetchall() + valid_fields = {row[1] for row in info} + if field not in valid_fields: + raise ValueError(f"Invalid URL field: {field}") + + row = db.execute(f"SELECT {field} FROM urls WHERE url = ?", (url,)).fetchone() + old_value = row[0] if row else None + + if old_value == new_value: + return False + + db.execute(f"UPDATE urls SET {field} = ? WHERE url = ?", (new_value, url)) + record_url_history(db, url, field, old_value, new_value, changed_by) + return True + + def set_url_latest_content(db, url, content_hash): """ Mark content_hash as the latest snapshot for url. diff --git a/web/main.py b/web/main.py index b6b1278..a7f0b44 100644 --- a/web/main.py +++ b/web/main.py @@ -17,6 +17,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 update_url_field from common.utils import is_valid, get_domain # Global variables @@ -309,7 +310,11 @@ def edit_detail(): classification = flask.request.form['class'] reason = flask.request.form['reason'] evaluated = "yes" if classification != "unclassified" else "no" - db.execute("UPDATE urls SET note = ?, classification = ?, classification_reason = ?, last_edit = ?, evaluated = ? WHERE url = ?", (note, classification, reason, user, evaluated, url)) + update_url_field(db, url, "note", note, changed_by=user) + update_url_field(db, url, "classification", classification, changed_by=user) + update_url_field(db, url, "classification_reason", reason, changed_by=user) + update_url_field(db, url, "evaluated", evaluated, changed_by=user) + update_url_field(db, url, "last_edit", user, changed_by=user) if classification == "malicious": back_propagation(db, url) return redirect(url_for("list_all", show=show)) @@ -345,15 +350,18 @@ def bulk_edit_action(): evaluated = "yes" if classification != "unclassified" else "no" urls_string = "('" + "', '".join(selected_urls) + "')" with SQLiteWrapper(config.db_path) as db: - if note: - db.execute(f"UPDATE urls SET note = ?, last_edit = ?, evaluated = ? WHERE url IN {urls_string}", (note, user, evaluated)) - if classification: - db.execute(f"UPDATE urls SET classification = ?, last_edit = ?, evaluated = ? WHERE url IN {urls_string}", (classification, user, evaluated)) - if classification_reason: - db.execute(f"UPDATE urls SET classification_reason = ?, last_edit = ?, evaluated = ? WHERE url IN {urls_string}", (classification_reason, user, evaluated)) + for target_url in selected_urls: + if note: + update_url_field(db, target_url, "note", note, changed_by=user) + if classification: + update_url_field(db, target_url, "classification", classification, changed_by=user) + if classification_reason: + update_url_field(db, target_url, "classification_reason", classification_reason, changed_by=user) + update_url_field(db, target_url, "evaluated", evaluated, changed_by=user) + update_url_field(db, target_url, "last_edit", user, changed_by=user) if classification == "malicious": - for url in selected_urls: - back_propagation(db, url) + for target_url in selected_urls: + back_propagation(db, target_url) return redirect(url_for("list_all")) From 73282bfeabcc8970bf968c67f7dad3dc3dd39fc1 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Fri, 24 Jul 2026 12:22:11 +0200 Subject: [PATCH 7/8] Update web interface for new data model - Add global search box and bulk URL textarea to list page - Show observations, content history, HTTP headers, change history on detail page - Add download blob and request-sandbox stub routes - Make MISP optional for dev/local runs --- web/main.py | 171 +++++++++++++++++++++++++++++++++--- web/templates/detail.html | 129 +++++++++++++++++++++++++-- web/templates/list_all.html | 59 ++++++++----- 3 files changed, 318 insertions(+), 41 deletions(-) diff --git a/web/main.py b/web/main.py index a7f0b44..6991664 100644 --- a/web/main.py +++ b/web/main.py @@ -7,6 +7,7 @@ import argparse import logging import base64 +import json from datetime import datetime, timezone from flask import Flask, jsonify, render_template, make_response, redirect, url_for @@ -19,6 +20,7 @@ from common.db import SQLiteWrapper from common.db_helpers import update_url_field from common.utils import is_valid, get_domain +from common.content_storage import load_content, storage_path # Global variables page = 1 @@ -44,8 +46,10 @@ config = Config(args.config) # Connect to MISP +misp = None try: - misp = PyMISP(config.misp_url, config.misp_key, config.misp_verify_cert) + if config.misp_url and not config.misp_url.startswith("!!! CHANGE"): + misp = PyMISP(config.misp_url, config.misp_key, config.misp_verify_cert) except Exception as e: print(f"Error: Cannot connect to MISP: {e}") @@ -67,6 +71,8 @@ def get_urlhaus_link(url): def get_misp_link(url_detail): + if misp is None: + return None if url_detail.reported == "yes": try: events = misp.search(controler="events", value=url_detail.url, type_attribute='url') @@ -179,18 +185,36 @@ def list_all(): except BadRequestKeyError: pass - # add new url + # add new urls (one or more, newline separated) try: - if (add_url := flask.request.form['add-url'].strip()) and is_valid(add_url): + if add_url_text := flask.request.form['add-url'].strip(): + added = 0 + skipped = 0 + failed = 0 with SQLiteWrapper(config.db_path) as db: t_now = datetime.now(timezone.utc).strftime('%Y-%m-%d') - in_db = db.execute("SELECT url, occurrences FROM urls WHERE url = ?", (add_url,)).fetchall() - if not in_db: - db.execute("INSERT INTO urls (url, first_seen, last_seen, domain) VALUES (?, ?, ?, ?)", (add_url, t_now, t_now, get_domain(add_url))) - db.execute("INSERT OR IGNORE INTO url_source (url, source) VALUES (?, ?)", (add_url, "Manual")) - adding = "success" - else: - adding = "in_db" + for add_url in add_url_text.splitlines(): + add_url = add_url.strip() + if not add_url: + continue + if not is_valid(add_url): + failed += 1 + continue + in_db = db.execute("SELECT url, occurrences FROM urls WHERE url = ?", (add_url,)).fetchall() + if not in_db: + db.execute("INSERT INTO urls (url, first_seen, last_seen, domain) VALUES (?, ?, ?, ?)", (add_url, t_now, t_now, get_domain(add_url))) + db.execute("INSERT OR IGNORE INTO url_source (url, source, observed_at) VALUES (?, ?, ?)", (add_url, "Manual", datetime.now(timezone.utc).isoformat())) + added += 1 + else: + skipped += 1 + if failed and not added and not skipped: + adding = "fail" + elif skipped and not added and not failed: + adding = "in_db" + elif added: + adding = "success" + else: + adding = "fail" else: adding = "fail" except BadRequestKeyError: @@ -268,7 +292,77 @@ def detail(): url_detail.src = [row[0] for row in db.execute("SELECT source FROM url_source WHERE url = ?", (url,)).fetchall()] url_detail.src_urls = db.execute("SELECT src_url FROM discovered_urls WHERE url = ?", (url_detail.url,)).fetchall() url_detail.contained_urls = db.execute("SELECT url FROM discovered_urls WHERE src_url = ?", (url,)).fetchall() - sessions = db.execute("SELECT sessions.session, sessions.idea_id FROM sessions JOIN url_session ON url_session.session=sessions.session_hash WHERE url_session.url = ?", (url,)).fetchall() + url_detail.observations = [ + { + "source": row[0], + "source_detail": row[1], + "origin_url": row[2], + "observed_at": row[3], + "idea_id": row[4], + "session_hash": row[5], + } + for row in db.execute( + "SELECT source, source_detail, origin_url, observed_at, idea_id, session_hash FROM url_source WHERE url = ? ORDER BY observed_at DESC", + (url,), + ).fetchall() + ] + url_detail.content_history = [ + { + "content_hash": row[0], + "first_seen": row[1], + "last_seen": row[2], + "is_latest": row[3], + "sha1": row[4], + "mime_type": row[5], + "content_size": row[6], + "http_status": row[7], + "downloaded_at": row[8], + } + for row in db.execute( + """ + SELECT cs.content_hash, uc.first_seen, uc.last_seen, uc.is_latest, + cs.sha1, cs.mime_type, cs.content_size, cs.http_status, cs.downloaded_at + FROM url_content uc + JOIN content_snapshot cs ON cs.content_hash = uc.content_hash + WHERE uc.url = ? + ORDER BY uc.first_seen DESC + """, + (url,), + ).fetchall() + ] + url_detail.latest_snapshot_headers = None + if url_detail.content_history: + latest_hash = url_detail.content_history[0]["content_hash"] + headers_row = db.execute( + "SELECT http_headers FROM content_snapshot WHERE content_hash = ?", + (latest_hash,), + ).fetchone() + if headers_row and headers_row[0]: + try: + url_detail.latest_snapshot_headers = json.dumps( + json.loads(headers_row[0]), indent=2, ensure_ascii=False + ) + except Exception: + url_detail.latest_snapshot_headers = headers_row[0] + url_detail.history = [ + { + "changed_at": row[0], + "field": row[1], + "old_value": row[2], + "new_value": row[3], + "changed_by": row[4], + } + for row in db.execute( + "SELECT changed_at, field, old_value, new_value, changed_by FROM url_history WHERE url = ? ORDER BY changed_at DESC", + (url,), + ).fetchall() + ] + sessions = [] + try: + sessions = db.execute("SELECT sessions.session, sessions.idea_id FROM sessions JOIN url_session ON url_session.session=sessions.session_hash WHERE url_session.url = ?", (url,)).fetchall() + except Exception: + # url_session may not exist in migrated databases yet + sessions = [] # count not active days inactive_for = 0 @@ -317,7 +411,7 @@ def edit_detail(): update_url_field(db, url, "last_edit", user, changed_by=user) if classification == "malicious": back_propagation(db, url) - return redirect(url_for("list_all", show=show)) + return redirect(url_for("detail", url=url, show=show)) url_list = db.execute("SELECT * FROM urls WHERE url = ? LIMIT 1", (url,)).fetchall() @@ -395,3 +489,56 @@ def api_url_stats(): "src": ", ".join([s[0] for s in url_sources]), } return make_response(jsonify(return_dict), 200) + + +@app.route('/download/', methods=['GET']) +def download_content(content_hash): + """Serve a stored content snapshot as a binary download.""" + import re + if not re.fullmatch(r"[0-9a-fA-F]{64}", content_hash): + return make_response(jsonify({'error': 'Invalid content hash'}), 400) + + base_dir = getattr(config, 'content_storage_path', os.path.join(os.path.dirname(config.db_path), 'content')) + try: + data = load_content(base_dir, content_hash) + except FileNotFoundError: + return make_response(jsonify({'error': 'Content not found'}), 404) + + response = make_response(data) + response.headers['Content-Type'] = 'application/octet-stream' + response.headers['Content-Disposition'] = f'attachment; filename="{content_hash}.blob"' + return response + + +@app.route('/request_sandbox', methods=['POST']) +def request_sandbox(): + """Stub endpoint to request sandbox analysis for a content snapshot.""" + user = get_user(flask.request.environ) + url = flask.request.form.get('url') or flask.request.args.get('url') + content_hash = flask.request.form.get('content_hash') + + if not url or not content_hash: + return make_response(jsonify({'error': 'Missing url or content_hash'}), 400) + + import re + if not re.fullmatch(r"[0-9a-fA-F]{64}", content_hash): + return make_response(jsonify({'error': 'Invalid content hash'}), 400) + + submitted_at = datetime.now(timezone.utc).isoformat() + with SQLiteWrapper(config.db_path) as db: + snapshot = db.execute( + "SELECT 1 FROM content_snapshot WHERE content_hash = ?", (content_hash,) + ).fetchone() + if not snapshot: + return make_response(jsonify({'error': 'Content snapshot not found'}), 404) + + db.execute( + """ + INSERT INTO sandbox_job + (content_hash, url, provider, status, submitted_at, requested_by) + VALUES (?, ?, ?, 'pending', ?, ?) + """, + (content_hash, url, 'anyrun', submitted_at, user), + ) + + return redirect(url_for('detail', url=url)) diff --git a/web/templates/detail.html b/web/templates/detail.html index bb4803d..563932f 100755 --- a/web/templates/detail.html +++ b/web/templates/detail.html @@ -1,9 +1,9 @@ {% extends "main.html" %} {% block content %} -
@@ -66,7 +66,7 @@ - Source + Sources {{ url.src | join(', ') }} @@ -194,7 +194,126 @@
-
+ + {% if url.observations %} +
+

Observations

+ + + + + + + + + + + + {% for obs in url.observations %} + + + + + + + + {% endfor %} + +
SourceDetailObserved atOrigin URLSession / IDEA
{{ obs.source or '-' }}{{ obs.source_detail or '-' }}{{ obs.observed_at or '-' }} + {% if obs.origin_url %} + {{ obs.origin_url }} + {% else %}-{% endif %} + + {% if obs.idea_id %} + + {% endif %} + {% if obs.session_hash %} + {{ obs.session_hash[:16] }}... + {% endif %} +
+
+ {% endif %} + + + {% if url.content_history %} +
+

Content history

+ + + + + + + + + + + + + + + + {% for entry in url.content_history %} + + + + + + + + + + + + {% endfor %} + +
First seenLast seenSHA-256SHA-1MIME typeSizeHTTP statusDownloaded atActions
{{ entry.first_seen or '-' }}{{ entry.last_seen or '-' }}{{ entry.content_hash[:16] }}...{{ entry.sha1[:16] if entry.sha1 else '-' }}...{{ entry.mime_type or '-' }}{{ entry.content_size if entry.content_size is not none else '-' }} B{{ entry.http_status or '-' }}{{ entry.downloaded_at or '-' }} + Download +
+ + +
+
+
+ + {% if url.latest_snapshot_headers %} +
+

Latest HTTP headers (click to toggle)

+ +
+ {% endif %} + {% endif %} + + + {% if url.history %} +
+

Change history

+ + + + + + + + + + + + {% for h in url.history %} + + + + + + + + {% endfor %} + +
Changed atFieldOld valueNew valueBy
{{ h.changed_at or '-' }}{{ h.field }}{{ h.old_value or '-' }}{{ h.new_value or '-' }}{{ h.changed_by or '-' }}
+
+ {% endif %} + +
{% if url.evaluated == 'yes' %}
diff --git a/web/templates/list_all.html b/web/templates/list_all.html index 37d3809..50871d0 100755 --- a/web/templates/list_all.html +++ b/web/templates/list_all.html @@ -16,51 +16,62 @@
{% endif %}
- + + + +
- -
+ +
- +
-
- +
-
- +
@@ -69,7 +80,7 @@
@@ -78,8 +89,8 @@
@@ -90,9 +101,9 @@
- - - + + +
From d1c767ba325c558bd8824233868712ce2495fdd0 Mon Sep 17 00:00:00 2001 From: Lukas Simonik Date: Mon, 27 Jul 2026 13:55:38 +0200 Subject: [PATCH 8/8] Adapt export scripts to new data model - evaluator2misp.py / evaluator2misp_bysource.py: - add latest_content_hash (sha256) and content_size to exported MISP objects - keep existing sha1 hash and source metadata - switch source aggregation to SQLite-compatible subquery to avoid DISTINCT aggregate errors - evaluator2urlhaus.py: - include sha256, sha1, file_mime_type and content_size in URLhaus submissions - still only send active malicious URLs, skipping already-blacklisted ones --- bin/evaluator2misp.py | 18 +++++++++++---- bin/evaluator2misp_bysource.py | 18 +++++++++++---- bin/evaluator2urlhaus.py | 40 ++++++++++++++++++++++++---------- 3 files changed, 57 insertions(+), 19 deletions(-) diff --git a/bin/evaluator2misp.py b/bin/evaluator2misp.py index e6a0d89..2a80e9b 100644 --- a/bin/evaluator2misp.py +++ b/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") diff --git a/bin/evaluator2misp_bysource.py b/bin/evaluator2misp_bysource.py index 532d39b..33b9e14 100644 --- a/bin/evaluator2misp_bysource.py +++ b/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") diff --git a/bin/evaluator2urlhaus.py b/bin/evaluator2urlhaus.py index f11f30e..158b55a 100644 --- a/bin/evaluator2urlhaus.py +++ b/bin/evaluator2urlhaus.py @@ -17,13 +17,17 @@ 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") @@ -31,22 +35,36 @@ def evaluator2urlhaus(): # 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):