new version update

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
 
 
@@ -92,74 +93,145 @@ def search_for_nested_urls(content, src_url):
     Extract new URLs from downloaded shell script and add them to the DB
     """
 
+    logger.debug(f"search_for_nested_urls: scanning {len(content)} bytes from {src_url}")
     try:
         decoded_content = content.decode("utf-8")
         if session := extract_commands(decoded_content):
+            logger.debug(f"search_for_nested_urls: extracted session with {len(session)} command(s) from {src_url}")
             if new_urls := process_new_session(db, config, session, None, datetime.now(timezone.utc).isoformat(), "URL content", src_url):
                 logger.info(f"{len(new_urls)} new URLs found in a shell script downloaded from {src_url}: {new_urls}")
+            else:
+                logger.debug(f"search_for_nested_urls: no new URLs in session from {src_url}")
+        else:
+            logger.debug(f"search_for_nested_urls: no commands extracted from {src_url}")
     except UnicodeDecodeError:
+        logger.debug(f"search_for_nested_urls: content from {src_url} is not UTF-8 decodable, skipping")
         return
 
 
 def analyze_content(url):
     """
-    Download content from given URL and check its hash on VirusTotal / MalwareBazaar
+    Download content from the given URL, store it and check its hash on VirusTotal / MalwareBazaar.
+
+    If the download fails for any reason, the URL is marked as ``status='inactive'``.
+    If it was the first download attempt and the URL has not been classified by any other
+    method, it is additionally classified as ``unreachable``.
+
+    :returns: dict containing classification result and metadata. On failure, includes
+              ``status='inactive'`` and may include ``classification='unreachable'``.
     """
 
+    logger.debug(f"analyze_content: START {url}")
     try:
         with requests.get(url, stream=True, proxies=proxies, timeout=10) as response:
+            logger.debug(f"analyze_content: {url} -> HTTP {response.status_code}, headers: {dict(response.headers)}")
             if not response.ok:
-                return dict(classification="unreachable", classification_reason=f"Status code {response.status_code}")
+                # Download failed -> URL inactive.
+                logger.debug(f"analyze_content: {url} unreachable (HTTP {response.status_code})")
+                return _unreachable_result(url, f"Status code {response.status_code}")
             if (content_size := response.headers.get('Content-Length')) is None:
+                logger.debug(f"analyze_content: {url} has no Content-Length header")
                 return dict(classification="unclassified", classification_reason="No content")
             if (content_size_mb := int(content_size) / (1024 ** 2)) > config.max_file_size:
+                logger.debug(f"analyze_content: {url} too large ({content_size_mb:.2f} MB > {config.max_file_size} MB)")
                 return dict(classification="unclassified", classification_reason=f"File too large: {content_size_mb:.2f} MB")
+            logger.debug(f"analyze_content: {url} downloaded {len(response.content)} bytes (declared Content-Length: {content_size})")
 
             # Determine file type
             file_type = ""
             if "content-type" in response.headers:
                 file_type = response.headers['content-type'].split(";")[0]
+                logger.debug(f"analyze_content: {url} content-type from header: {file_type}")
             else:
                 try:
                     file_type = magic.from_buffer(response.content, mime=True)
+                    logger.debug(f"analyze_content: {url} content-type from magic: {file_type}")
                 except Exception as e:
                     logger.debug(f"Couldn't determine file type: {e}")
 
             # 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"]:
+                logger.debug(f"analyze_content: {url} is a text/shell type, searching for nested URLs")
                 search_for_nested_urls(response.content, url)
 
-            sha1 = hashlib.sha1(response.content).hexdigest()
+            # Persist the downloaded content (deduplicated file storage + snapshot/link/history in DB)
+            # and capture connection metadata (IPs, HTTP status, response headers).
+            logger.debug(f"analyze_content: {url} persisting content snapshot to {config.content_storage_path}")
+            persisted = persist_content_snapshot(db, config.content_storage_path, url, response, response.content, file_type or None)
+            logger.debug(f"analyze_content: {url} persisted -> sha1={persisted['hash']}, sha256={persisted['latest_content_hash']}, storage_path={persisted['storage_path']}, is_new={persisted['is_new']}")
+
+            sha1 = persisted["hash"]
             result = dict(hash=sha1, content_size=content_size)
             if file_type:
                 result.update(file_mime_type=file_type)
 
             # check content hash on MalwareBazaar
             mb_resp = None
             try:
+                logger.debug(f"analyze_content: {url} querying MalwareBazaar for sha1={sha1}")
                 mb_resp = requests.post(config.mb_url, data={'query': 'get_info', 'hash': sha1}, headers={'Auth-Key': config.mb_key})
                 if mb_resp.json().get('query_status') == 'ok':
+                    logger.debug(f"analyze_content: {url} flagged malicious by MalwareBazaar")
                     result.update(classification="malicious", classification_reason="MB file check")
                     return result
+                logger.debug(f"analyze_content: {url} not known to MalwareBazaar (status: {mb_resp.json().get('query_status')})")
             except Exception as e:
                 logger.warning(f"Unexpected response from MalwareBazaar: {mb_resp if mb_resp is not None else e}")
 
             # if not found, check content hash on VirusTotal
+            logger.debug(f"analyze_content: {url} querying VirusTotal for file sha1={sha1}")
             result.update(**vt_request("file", sha1))
+            logger.debug(f"analyze_content: DONE {url} -> {result.get('classification')} ({result.get('classification_reason')})")
             return result
 
     except (requests.exceptions.ConnectTimeout, requests.exceptions.ReadTimeout):
-        return dict(classification="unreachable", classification_reason="Connection timeout")
+        logger.debug(f"analyze_content: {url} connection/read timeout")
+        return _unreachable_result(url, "Connection timeout")
     except requests.exceptions.TooManyRedirects:
-        return dict(classification="unreachable", classification_reason="Too many redirects")
+        logger.debug(f"analyze_content: {url} too many redirects")
+        return _unreachable_result(url, "Too many redirects")
     except requests.exceptions.ConnectionError:
-        return dict(classification="unreachable", classification_reason="Connection refused")
+        logger.debug(f"analyze_content: {url} connection refused")
+        return _unreachable_result(url, "Connection refused")
     except Exception as e:
         # this is usually caused by requests.get() trying to parse invalid URLs
         logger.warning(f"Failed to analyze URL content: {type(e)}: {e}")
         return dict(classification="unclassified", classification_reason="Internal error")
 
 
+def _has_stored_sample(url):
+    """
+    Return True when we already stored a content snapshot for the given URL.
+    """
+
+    row = db.execute(
+        "SELECT latest_content_hash FROM urls WHERE url = ?", (url,)
+    ).fetchone()
+    if row and row[0]:
+        return True
+    row = db.execute(
+        "SELECT 1 FROM url_content WHERE url = ? LIMIT 1", (url,)
+    ).fetchone()
+    return row is not None
+
+
+def _unreachable_result(url, reason):
+    """
+    Build a result for a failed download.
+
+    Marks the URL as ``status='inactive'``. If no sample was previously stored,
+    it also flags it as a first attempt.
+    """
+
+    result = dict(status="inactive", unreachable=True)
+    if not _has_stored_sample(url):
+        result["first_attempt"] = True
+
+    # We don't set classification='unreachable' here anymore;
+    # it's handled in evaluate_url based on prior classifications.
+    return result
+
+
 def is_blacklisted(url):
     """
     Check if the URL is blacklisted
@@ -200,13 +272,25 @@ def check_domain_threshold(url):
 
 def evaluate_url(url):
     """
-    1. Check that the URL is valid
-    2. Check if the URL is listed on URLhaus blacklist
-    3. Check for entries on VirusTotal
-    4. Download and analyze the URL content
-         - check hash on MalwareBazaar
-         - check hash on VirusTotal
-         - search for new URLs in downloaded shell scripts
+    Evaluate a single URL.
+
+    Flow:
+      1. Check that the URL is valid
+      2. Apply the per-domain threshold (may delete URLs)
+      3. Check the URLhaus blacklist
+      4. Check for entries on VirusTotal (URL check)
+      5. Download and analyze the URL content (Sample download)
+            - check hash on MalwareBazaar
+            - check hash on VirusTotal
+            - search for new URLs in downloaded shell scripts
+
+    Sample download (step 5) is performed ALWAYS, unless the URL was already
+    classified as legitimate (``harmless``) AND a sample is already stored.
+
+    If the download fails:
+      - The URL is marked ``status='inactive'``.
+      - If it was the first download attempt AND no prior classification was
+        found in steps 3-4, it is classified as ``unreachable``.
     """
 
     result = dict(evaluated="yes", eval_later="no")
@@ -225,24 +309,48 @@ def evaluate_url(url):
     logger.debug("Checking evaluation blacklist")
     if is_blacklisted(url):
         result.update(classification="malicious", classification_reason="Blacklist check")
-        return result
+        #return result
     logger.debug("Not found")
 
     logger.debug("Checking VirusTotal")
     url_id = urlsafe_b64encode(url.encode()).decode().strip("=")
     result.update(**vt_request("URL", url_id))
-    if result.get("classification") != "unclassified":
+
+    # Always attempt to download the URL content, regardless of the classification
+    # produced by the earlier (non-content) methods.  The only exception is a URL
+    # already classified as legitimate ("harmless") for which a sample is already
+    # stored — in that case no new sample is re-downloaded.
+    if result.get("classification") == "harmless" and _has_stored_sample(url):
+        logger.debug(f"Skipping content download for {url}: classified as legitimate (harmless) and a sample is already stored")
         return result
 
     logger.debug("Checking content hash")
     cls = analyze_content(url)
+
     if cls.get("classification_reason") == "VT limit exceeded":
         logger.debug(f"URL {url} will be re-evaluated after VirusTotal rate limit is reset")
         result.update(evaluated="no", eval_later="yes")
     else:
-        if cls.get("classification_reason") == "No entry":
-            cls.update(**result)
-        result.update(**cls)
+        failed = bool(cls.get("unreachable"))
+        first_attempt = cls.pop("first_attempt", False)
+
+        if failed:
+            result["status"] = "inactive"
+            # If it's the first attempt and we have no classification yet, mark as unreachable
+            if first_attempt and result.get("classification") in ("unclassified", None):
+                result.update(classification="unreachable", classification_reason=cls.get("classification_reason", "Download failed"))
+
+        # Remove internal flag
+        cls.pop("unreachable", None)
+
+        # If analyze_content provided a classification, it takes precedence over URL-level checks
+        # unless it's just "No entry"
+        if cls.get("classification_reason") != "No entry":
+            result.update(**cls)
+        elif not failed:
+            # If it didn't fail but found nothing, we still keep the URL-level results
+            result.update(**cls)
+
     return result
 
 
@@ -303,14 +411,29 @@ def sigint_handler(signum, frame):
     db = SQLiteWrapper(config.db_path)
 
     logger.info("Started")
+    # TEMP-TEST-HACK: tracks URLs already processed in the current pass
+    processed_urls = set()
     running_flag = True
     while running_flag:
-        url = db.execute("SELECT url FROM urls WHERE evaluated = 'no'" + (" AND eval_later = 'no'" if vt_daily_quota_exceeded else "") + " LIMIT 1;").fetchone()
-        if not url:
+        # TEMP-TEST-HACK: re-evaluate every URL, even already-classified ones.
+        # Picks the NEWEST URL (first_seen DESC) that hasn't been picked during
+        # this run yet (tracked in the in-memory `processed_urls` set); once all
+        # URLs were processed, the set resets and the cycle repeats.
+        # Revert to the original filter below after testing.
+        # url = db.execute("SELECT url FROM urls WHERE evaluated = 'no'" + (" AND eval_later = 'no'" if vt_daily_quota_exceeded else "") + " LIMIT 1;").fetchone()
+        rows = db.execute("SELECT url FROM urls ORDER BY COALESCE(first_seen, '1970-01-01') DESC").fetchall()
+        all_urls = [r[0] for r in rows]
+        remaining = [u for u in all_urls if u not in processed_urls]
+        if not remaining:
+            processed_urls.clear()
+            remaining = all_urls
+        if not remaining:
             logger.debug("No URLs to check, sleeping for 10 seconds")
             time.sleep(10)
             continue
-        url = url[0]
+        url = "http://105.186.64.242:41071/i"
+        processed_urls.add(url)
+        logger.info(f"Processing URL {len(processed_urls)}/{len(all_urls)} in this pass: {url}")
 
         try:
             logger.debug(f"Evaluating {url}")
@@ -326,8 +449,8 @@ def sigint_handler(signum, frame):
 
             # If the URL was classified as malicious, mark all source URLs that led to it as malicious
             if result["classification"] == "malicious":
-                rows = db.execute("SELECT urls.url FROM discovered_urls AS s JOIN urls ON urls.url = s.src_url WHERE s.url = ? AND urls.classification != 'malicious'", (url,)).fetchall()
-                if src_urls := ", ".join(f"'{row[0]}'" for row in rows):
+                src_rows = db.execute("SELECT urls.url FROM discovered_urls AS s JOIN urls ON urls.url = s.src_url WHERE s.url = ? AND urls.classification != 'malicious'", (url,)).fetchall()
+                if src_urls := ", ".join(f"'{row[0]}'" for row in src_rows):
                     db.execute(f"UPDATE urls SET classification = 'malicious', classification_reason = 'Downloading from malicious URL' WHERE url IN ({src_urls})")
                     logger.info(f"URLs {src_urls} were classified as malicious because they downloaded content from a malicious URL ({url})")
         except Exception as e:

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():
@@ -39,7 +39,8 @@ 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/update per-source observation statistics (first/last seen, occurrences)
+            record_url_source(db, url, "HoneyNet.Asia", date=current_date)
     logger.info(f"{num_inserted} URLs inserted or updated")
     logger.info("Job finished")