mirror of
https://github.com/reconurge/flowsint.git
synced 2026-07-21 06:44:58 -05:00
feat(types): detect MD5/SHA1/SHA256 file hashes on import
TXT imports of file hashes were detected as Username, because File.detect always returned False and Username.detect matches any 3-80 char alphanumeric string. Hashes therefore couldn't be used as File entities to pivot from (VirusTotal, MalwareBazaar, sandboxes), per #45. - File.detect now recognizes 32/40/64-char hex strings (MD5/SHA1/SHA256), and File.from_string stores the value in the matching hash_* field (normalized to lowercase) while keeping it as the filename/label. - Username.detect now explicitly rejects hash-shaped strings, so hash detection is correct regardless of type-registration order (File is already registered before Username, but this removes the implicit dependency). Tests cover detection of each hash type (case/whitespace), rejection of non-hashes, from_string field population, the Username regression, and end-to-end detect_type() resolution to File. Closes #45
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
from flowsint_types import Domain, Ip, Phone
|
||||
from flowsint_types import Domain, File, Ip, Phone
|
||||
|
||||
from flowsint_core.imports import detect_type
|
||||
|
||||
@@ -40,6 +40,20 @@ def test_detection_ips():
|
||||
assert results[1].address == "12.43.23.12"
|
||||
|
||||
|
||||
def test_detection_file_hashes():
|
||||
hashes = {
|
||||
"d41d8cd98f00b204e9800998ecf8427e": "hash_md5",
|
||||
"da39a3ee5e6b4b0d3255bfef95601890afd80709": "hash_sha1",
|
||||
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855": "hash_sha256",
|
||||
}
|
||||
for value, field in hashes.items():
|
||||
DetectedType = detect_type(value)
|
||||
assert DetectedType is File, f"{value} should be detected as File"
|
||||
obj = DetectedType.from_string(value)
|
||||
assert isinstance(obj, File)
|
||||
assert getattr(obj, field) == value
|
||||
|
||||
|
||||
def test_detection_mixed():
|
||||
raw_obj = ["240.123.123.234", "my.domain.io", "+33632233223"]
|
||||
results = []
|
||||
|
||||
@@ -1,9 +1,27 @@
|
||||
import re
|
||||
|
||||
from pydantic import Field, model_validator
|
||||
from typing import Optional, List, Self
|
||||
from typing import Optional, Self
|
||||
|
||||
from .flowsint_base import FlowsintType
|
||||
from .registry import flowsint_type
|
||||
|
||||
# Map of hex-hash length -> the File field that should hold it.
|
||||
_HASH_LENGTH_TO_FIELD = {
|
||||
32: "hash_md5",
|
||||
40: "hash_sha1",
|
||||
64: "hash_sha256",
|
||||
}
|
||||
_HASH_RE = re.compile(r"^[0-9a-fA-F]+$")
|
||||
|
||||
|
||||
def _hash_field_for(value: str) -> Optional[str]:
|
||||
"""Return the File hash field name for an MD5/SHA1/SHA256 hex string, else None."""
|
||||
value = value.strip()
|
||||
if not _HASH_RE.match(value):
|
||||
return None
|
||||
return _HASH_LENGTH_TO_FIELD.get(len(value))
|
||||
|
||||
|
||||
@flowsint_type
|
||||
class File(FlowsintType):
|
||||
@@ -73,10 +91,22 @@ class File(FlowsintType):
|
||||
|
||||
@classmethod
|
||||
def from_string(cls, line: str):
|
||||
"""Parse a file from a raw string."""
|
||||
return cls(filename=line.strip())
|
||||
"""Parse a file from a raw string.
|
||||
|
||||
If the value is an MD5/SHA1/SHA256 hash, store it in the matching
|
||||
hash field as well as the filename (so the node keeps a label).
|
||||
"""
|
||||
line = line.strip()
|
||||
hash_field = _hash_field_for(line)
|
||||
if hash_field:
|
||||
return cls(filename=line, **{hash_field: line.lower()})
|
||||
return cls(filename=line)
|
||||
|
||||
@classmethod
|
||||
def detect(cls, line: str) -> bool:
|
||||
"""File cannot be reliably detected from a single line of text."""
|
||||
return False
|
||||
"""Detect a file hash (MD5 / SHA1 / SHA256) from a single line.
|
||||
|
||||
Only hashes are auto-detected; arbitrary filenames are too ambiguous
|
||||
to detect reliably, so non-hash values still return False.
|
||||
"""
|
||||
return _hash_field_for(line) is not None
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import re
|
||||
from typing import Any, Optional, Self
|
||||
from typing import Optional, Self
|
||||
|
||||
from pydantic import Field, field_validator, model_validator
|
||||
|
||||
@@ -60,6 +60,11 @@ class Username(FlowsintType):
|
||||
if not line:
|
||||
return False
|
||||
|
||||
# Don't claim file hashes (MD5/SHA1/SHA256) — those are File entities.
|
||||
# This keeps hash detection independent of type-registration order.
|
||||
if len(line) in (32, 40, 64) and re.fullmatch(r"[0-9a-fA-F]+", line):
|
||||
return False
|
||||
|
||||
# Username pattern: 3-80 characters, only letters, numbers, underscores, hyphens
|
||||
# Note: This is intentionally restrictive to avoid false positives
|
||||
return bool(re.match(r"^[a-zA-Z0-9_-]{3,80}$", line))
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Tests for File hash detection (issue #45)."""
|
||||
|
||||
from flowsint_types import File, Username
|
||||
|
||||
MD5 = "d41d8cd98f00b204e9800998ecf8427e" # 32 hex
|
||||
SHA1 = "da39a3ee5e6b4b0d3255bfef95601890afd80709" # 40 hex
|
||||
SHA256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" # 64 hex
|
||||
|
||||
|
||||
def test_detect_md5():
|
||||
assert File.detect(MD5) is True
|
||||
|
||||
|
||||
def test_detect_sha1():
|
||||
assert File.detect(SHA1) is True
|
||||
|
||||
|
||||
def test_detect_sha256():
|
||||
assert File.detect(SHA256) is True
|
||||
|
||||
|
||||
def test_detect_is_case_insensitive():
|
||||
assert File.detect(SHA256.upper()) is True
|
||||
|
||||
|
||||
def test_detect_strips_whitespace():
|
||||
assert File.detect(f" {MD5} ") is True
|
||||
|
||||
|
||||
def test_detect_rejects_non_hash():
|
||||
# Plain filename, wrong length, and non-hex strings are not detected.
|
||||
assert File.detect("report.pdf") is False
|
||||
assert File.detect("john_doe") is False
|
||||
assert File.detect("g" * 32) is False # right length, not hex
|
||||
assert File.detect("abc123") is False # hex but not a hash length
|
||||
|
||||
|
||||
def test_from_string_populates_hash_fields():
|
||||
assert File.from_string(MD5).hash_md5 == MD5
|
||||
assert File.from_string(SHA1).hash_sha1 == SHA1
|
||||
assert File.from_string(SHA256).hash_sha256 == SHA256
|
||||
|
||||
|
||||
def test_from_string_normalizes_hash_to_lowercase():
|
||||
f = File.from_string(SHA256.upper())
|
||||
assert f.hash_sha256 == SHA256 # stored lowercase
|
||||
assert f.filename == SHA256.upper() # original kept as label
|
||||
|
||||
|
||||
def test_from_string_plain_filename_unchanged():
|
||||
f = File.from_string("evil.exe")
|
||||
assert f.filename == "evil.exe"
|
||||
assert f.hash_md5 is None
|
||||
assert f.hash_sha1 is None
|
||||
assert f.hash_sha256 is None
|
||||
|
||||
|
||||
def test_username_no_longer_claims_hashes():
|
||||
# Regression: hashes used to be mis-detected as usernames.
|
||||
assert Username.detect(MD5) is False
|
||||
assert Username.detect(SHA1) is False
|
||||
assert Username.detect(SHA256) is False
|
||||
# Real usernames still detected.
|
||||
assert Username.detect("john_doe") is True
|
||||
Reference in New Issue
Block a user