venus/CTF-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adding more features inclided steg analysis, decoding engines, and more

Browse Source 
tests
This commit is contained in:
venus
2026-07-19 00:13:13 -05:00
parent c48e4343dc
commit cd7ef151c4
20 changed files with 1360 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

440
src/ctf/forensics.py
View File

@@ -73,6 +73,12 @@ class FileMetadata:
# EXIF Data
exif_data: Dict[str, Any] = field(default_factory=dict)
# Physical Image Data
physical_data: Dict[str, Any] = field(default_factory=dict)
# Decoded metadata hints
decoded_hints: Dict[str, Dict[str, str]] = field(default_factory=dict)
# JPEG/PNG Comments
comment: str = ""
# }}}
@@ -158,6 +164,39 @@ def get_metadata(path: Path) -> FileMetadata:
exif_data = get_exif(p)
comment = get_comment(p)
physical_data = {}
try:
with open(p, "rb") as f:
file_bytes = f.read()
if file_bytes.startswith(b"\xff\xd8\xff"):
physical_data = get_jpeg_physical(file_bytes)
elif file_bytes.startswith(b"\x89PNG\r\n\x1a\n"):
physical_data = get_png_physical(file_bytes)
elif file_bytes.startswith(b"GIF87a") or file_bytes.startswith(b"GIF89a"):
physical_data = get_gif_physical(file_bytes)
except Exception:
pass
decoded_hints = {}
try:
from ctf.decoding import try_decode_metadata
if comment:
dec = try_decode_metadata(comment)
if dec:
decoded_hints["Comment"] = dec
for tag, val in exif_data.items():
if isinstance(val, str):
dec = try_decode_metadata(val)
if dec:
decoded_hints[f"EXIF:{tag}"] = dec
for attr, val in extended_attributes.items():
if isinstance(val, str):
dec = try_decode_metadata(val)
if dec:
decoded_hints[f"xattr:{attr}"] = dec
except Exception:
pass
return FileMetadata(
filename=p.name,
@@ -177,10 +216,13 @@ def get_metadata(path: Path) -> FileMetadata:
device=device,
extended_attributes=extended_attributes,
exif_data=exif_data,
physical_data=physical_data,
decoded_hints=decoded_hints,
comment=comment
)
# }}}
# {{{ EXIF tags
EXIF_TAGS = {
0x010e: "ImageDescription",
0x010f: "Make",
@@ -218,7 +260,7 @@ EXIF_TAGS = {
0xa403: "WhiteBalance",
0xa405: "FocalLengthIn35mmFilm",
0xa406: "SceneCaptureType",
}
}# }}}
# {{{ parse_tiff
def parse_tiff(data: bytes) -> Dict[str, Any]:
@@ -396,6 +438,20 @@ def get_exif(path: Path) -> Dict[str, Any]:
except Exception:
pass
# 3. Parse IPTC (Photoshop APP13) if JPEG
if data.startswith(b"\xff\xd8\xff"):
try:
tags.update(extract_jpeg_iptc(data))
except Exception:
pass
# 4. Parse PNG text chunks if PNG
if data.startswith(b"\x89PNG\r\n\x1a\n"):
try:
tags.update(parse_png_text_chunks(data))
except Exception:
pass
return tags
# }}}
@@ -425,7 +481,7 @@ def extract_jpeg_comment(data: bytes) -> str:
# {{{ get_comment
def get_comment(path: Path) -> str:
"""Reads file, checks headers, and extracts JPEG COM comments."""
"""Reads file, checks headers, and extracts JPEG COM / GIF comments."""
try:
with open(path, "rb") as f:
data = f.read()
@@ -434,16 +490,21 @@ def get_comment(path: Path) -> str:
if data.startswith(b"\xff\xd8\xff"):
return extract_jpeg_comment(data)
elif data.startswith(b"GIF87a") or data.startswith(b"GIF89a"):
return extract_gif_comments(data)
return ""
# }}}
#{{{ NS_map
NS_MAP = {
"http://www.w3.org/1999/02/22-rdf-syntax-ns#": "rdf",
"http://creativecommons.org/ns#": "cc",
"http://purl.org/dc/elements/1.1/": "dc",
"http://ns.adobe.com/xap/1.0/": "xmp",
"http://ns.adobe.com/exif/1.0/": "exif",
"http://ns.adobe.com/tiff/1.0/": "tiff",
}
#}}}
# {{{ parse_xmp
def parse_xmp(xmp_str: str) -> Dict[str, Any]:
@@ -470,7 +531,13 @@ def parse_xmp(xmp_str: str) -> Dict[str, Any]:
for elem in root.iter():
clean_tag = get_clean_name(elem.tag)
if clean_tag in ("rdf:RDF", "rdf:Description", "x:xmpmeta"):
if clean_tag == "rdf:Description":
for k, v in elem.attrib.items():
clean_k = get_clean_name(k)
if clean_k not in ("rdf:about", "rdf:Description"):
metadata[clean_k] = v
continue
if clean_tag in ("rdf:RDF", "x:xmpmeta"):
continue
val = None
@@ -482,7 +549,7 @@ def parse_xmp(xmp_str: str) -> Dict[str, Any]:
if "rdf:resource" in attribs:
val = attribs["rdf:resource"]
elif attribs:
val = ", ".join(f"{k}={v}" for k, v in attribs.items())
val = ", ".join(f"{k} | {v}" if v else k for k, v in sorted(attribs.items()))
if val:
metadata[clean_tag] = val
@@ -516,3 +583,368 @@ def extract_jpeg_xmp(data: bytes) -> str:
return ""
# }}}
# {{{ get_jpeg_physical
def get_jpeg_physical(data: bytes) -> Dict[str, Any]:
"""Parses JPEG APP0 and SOF segments for physical dimensions and metadata."""
physical = {}
if not data.startswith(b"\xff\xd8\xff"):
return physical
idx = 2
width = None
height = None
encoding_process = None
jfif_version = None
while idx < len(data) - 4:
if data[idx] == 0xff:
marker = data[idx+1]
if marker == 0xd9: # EOI
break
if marker in (0xd8, 0xd9, 0x00) or 0xd0 <= marker <= 0xd7:
idx += 2
continue
length = struct.unpack(">H", data[idx+2:idx+4])[0]
segment_data = data[idx+4:idx+4+length-2]
# APP0 (JFIF)
if marker == 0xe0 and segment_data.startswith(b"JFIF\x00"):
if len(segment_data) >= 7:
major = segment_data[5]
minor = segment_data[6]
jfif_version = f"{major}.{minor:02d}"
physical["JFIF Version"] = jfif_version
# SOF markers
elif marker in (0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb, 0xcd, 0xce, 0xcf):
if len(segment_data) >= 5:
precision = segment_data[0]
height = struct.unpack(">H", segment_data[1:3])[0]
width = struct.unpack(">H", segment_data[3:5])[0]
SOF_MARKERS = {
0xc0: "Baseline DCT, Huffman coding",
0xc1: "Extended Sequential DCT, Huffman coding",
0xc2: "Progressive DCT, Huffman coding",
0xc3: "Lossless, Huffman coding",
0xc5: "Differential Sequential DCT, Huffman coding",
0xc6: "Differential Progressive DCT, Huffman coding",
0xc7: "Differential Lossless, Huffman coding",
0xc9: "Extended Sequential DCT, Arithmetic coding",
0xca: "Progressive DCT, Arithmetic coding",
0xcb: "Lossless, Arithmetic coding",
0xcd: "Differential Sequential DCT, Arithmetic coding",
0xce: "Differential Progressive DCT, Arithmetic coding",
0xcf: "Differential Lossless, Arithmetic coding",
}
encoding_process = SOF_MARKERS.get(marker, "Unknown")
idx += 2 + length
else:
idx += 1
if width is not None and height is not None:
physical["Image Size"] = f"{width}x{height}"
mp = (width * height) / 1000000.0
physical["Megapixels"] = f"{mp:.1f}"
if encoding_process is not None:
physical["Encoding Process"] = encoding_process
return physical
# }}}
# {{{ get_png_physical
def get_png_physical(data: bytes) -> Dict[str, Any]:
"""Parses PNG IHDR and pHYs chunks for physical properties."""
physical = {}
if not data.startswith(b"\x89PNG\r\n\x1a\n"):
return physical
idx = 8
while idx < len(data) - 8:
length = struct.unpack(">I", data[idx:idx+4])[0]
chunk_type = data[idx+4:idx+8]
chunk_data = data[idx+8:idx+8+length]
if chunk_type == b"IHDR":
if len(chunk_data) >= 13:
width = struct.unpack(">I", chunk_data[0:4])[0]
height = struct.unpack(">I", chunk_data[4:8])[0]
bit_depth = chunk_data[8]
color_type = chunk_data[9]
compression = chunk_data[10]
filter_method = chunk_data[11]
interlace = chunk_data[12]
physical["Image Size"] = f"{width}x{height}"
mp = (width * height) / 1000000.0
physical["Megapixels"] = f"{mp:.1f}"
physical["Bit Depth"] = f"{bit_depth} bits/sample"
color_types = {
0: "Grayscale",
2: "Truecolor",
3: "Indexed-color",
4: "Grayscale with Alpha",
6: "Truecolor with Alpha"
}
physical["Color Type"] = color_types.get(color_type, f"Unknown ({color_type})")
if compression == 0:
physical["Encoding Process"] = "Deflate/Inflate"
else:
physical["Encoding Process"] = f"Unknown compression ({compression})"
interlace_methods = {
0: "Noninterlaced",
1: "Adam7 Interlace"
}
physical["Interlace Method"] = interlace_methods.get(interlace, f"Unknown ({interlace})")
elif chunk_type == b"pHYs":
if len(chunk_data) >= 9:
x_res = struct.unpack(">I", chunk_data[0:4])[0]
y_res = struct.unpack(">I", chunk_data[4:8])[0]
unit = chunk_data[8]
unit_str = " meters" if unit == 1 else " (unknown unit)"
physical["Pixels Per Unit X"] = f"{x_res}{unit_str}"
physical["Pixels Per Unit Y"] = f"{y_res}{unit_str}"
elif chunk_type == b"IEND":
break
idx += 12 + length
return physical
# }}}
# {{{ get_gif_physical
def get_gif_physical(data: bytes) -> Dict[str, Any]:
"""Parses GIF logical screen descriptor for physical size."""
physical = {}
if not (data.startswith(b"GIF87a") or data.startswith(b"GIF89a")):
return physical
if len(data) >= 10:
width = struct.unpack("<H", data[6:8])[0]
height = struct.unpack("<H", data[8:10])[0]
physical["Image Size"] = f"{width}x{height}"
mp = (width * height) / 1000000.0
physical["Megapixels"] = f"{mp:.1f}"
physical["Encoding Process"] = "LZW"
return physical
# }}}
# {{{ extract_jpeg_iptc
def extract_jpeg_iptc(data: bytes) -> Dict[str, str]:
"""Extracts IPTC/NAA metadata (Record 2) from Photoshop APP13 segments."""
iptc_metadata = {}
if not data.startswith(b"\xff\xd8\xff"):
return iptc_metadata
IPTC_TAGS = {
5: "ObjectName",
25: "Keywords",
40: "SpecialInstructions",
80: "By-line",
85: "By-lineTitle",
90: "City",
95: "Province-State",
101: "Country-PrimaryLocationName",
105: "Headline",
110: "Credit",
115: "Source",
116: "CopyrightNotice",
120: "Caption-Abstract",
122: "Writer-Editor"
}
idx = 2
while idx < len(data) - 4:
if data[idx] == 0xff:
marker = data[idx+1]
if marker == 0xd9: # EOI
break
if marker in (0xd8, 0xd9, 0x00) or 0xd0 <= marker <= 0xd7:
idx += 2
continue
length = struct.unpack(">H", data[idx+2:idx+4])[0]
if marker == 0xed: # APP13 Photoshop block
app13_data = data[idx+4:idx+4+length-2]
if app13_data.startswith(b"Photoshop 3.0\x00"):
offset = 14
while offset < len(app13_data) - 12:
if app13_data[offset:offset+4] == b"8BIM":
offset += 4
irb_id = app13_data[offset:offset+2]
offset += 2
name_len = app13_data[offset]
offset += 1
name = app13_data[offset:offset+name_len]
offset += name_len
if (name_len + 1) % 2 != 0:
offset += 1
if offset + 4 > len(app13_data):
break
size = struct.unpack(">I", app13_data[offset:offset+4])[0]
offset += 4
if offset + size > len(app13_data):
break
irb_data = app13_data[offset:offset+size]
offset += size
if size % 2 != 0:
offset += 1
if irb_id == b"\x04\x04":
iptc_offset = 0
while iptc_offset < len(irb_data) - 5:
if irb_data[iptc_offset] == 0x1c:
record = irb_data[iptc_offset+1]
dataset = irb_data[iptc_offset+2]
ds_size = struct.unpack(">H", irb_data[iptc_offset+3:iptc_offset+5])[0]
iptc_offset += 5
if iptc_offset + ds_size <= len(irb_data):
ds_data = irb_data[iptc_offset:iptc_offset+ds_size]
iptc_offset += ds_size
if record == 2:
tag_name = IPTC_TAGS.get(dataset, f"IPTC_2_{dataset}")
val = ds_data.decode("utf-8", errors="ignore")
iptc_metadata[tag_name] = val
else:
break
else:
iptc_offset += 1
idx += 2 + length
else:
idx += 1
return iptc_metadata
# }}}
# {{{ parse_png_text_chunks
def parse_png_text_chunks(data: bytes) -> Dict[str, str]:
"""Decompresses and extracts standard tEXt/zTXt/iTXt PNG text metadata."""
import zlib
text_metadata = {}
if not data.startswith(b"\x89PNG\r\n\x1a\n"):
return text_metadata
idx = 8
while idx < len(data) - 8:
length = struct.unpack(">I", data[idx:idx+4])[0]
chunk_type = data[idx+4:idx+8]
chunk_data = data[idx+8:idx+8+length]
if chunk_type == b"tEXt":
parts = chunk_data.split(b"\x00", 1)
if len(parts) == 2:
key = parts[0].decode("utf-8", errors="ignore")
val = parts[1].decode("utf-8", errors="ignore")
text_metadata[key] = val
elif chunk_type == b"zTXt":
parts = chunk_data.split(b"\x00", 1)
if len(parts) == 2:
key = parts[0].decode("utf-8", errors="ignore")
remaining = parts[1]
if len(remaining) > 1:
comp_method = remaining[0]
comp_text = remaining[1:]
if comp_method == 0:
try:
val = zlib.decompress(comp_text).decode("utf-8", errors="ignore")
text_metadata[key] = val
except Exception:
pass
elif chunk_type == b"iTXt":
parts = chunk_data.split(b"\x00", 1)
if len(parts) == 2:
key = parts[0].decode("utf-8", errors="ignore")
remaining = parts[1]
if len(remaining) >= 2:
comp_flag = remaining[0]
comp_method = remaining[1]
rem = remaining[2:]
parts2 = rem.split(b"\x00", 1)
if len(parts2) == 2:
lang_tag = parts2[0].decode("utf-8", errors="ignore")
rem = parts2[1]
parts3 = rem.split(b"\x00", 1)
if len(parts3) == 2:
trans_key = parts3[0].decode("utf-8", errors="ignore")
text_bytes = parts3[1]
if comp_flag == 1 and comp_method == 0:
try:
val = zlib.decompress(text_bytes).decode("utf-8", errors="ignore")
text_metadata[key] = val
except Exception:
pass
else:
val = text_bytes.decode("utf-8", errors="ignore")
text_metadata[key] = val
elif chunk_type == b"IEND":
break
idx += 12 + length
return text_metadata
# }}}
# {{{ extract_gif_comments
def extract_gif_comments(data: bytes) -> str:
"""Extracts comments from GIF comment extensions (\x21\xfe) sequentially."""
if not (data.startswith(b"GIF87a") or data.startswith(b"GIF89a")):
return ""
comments = []
idx = 6
if idx + 7 > len(data):
return ""
packed_byte = data[idx+4]
global_color_table_present = bool(packed_byte & 0x80)
global_color_table_size = 2 ** ((packed_byte & 0x07) + 1)
idx += 7
if global_color_table_present:
idx += 3 * global_color_table_size
while idx < len(data) - 2:
intro = data[idx]
if intro == 0x21: # Extension Introducer
ext_label = data[idx+1]
idx += 2
if ext_label == 0xfe: # Comment Extension
comment_parts = []
while idx < len(data):
block_len = data[idx]
idx += 1
if block_len == 0:
break
if idx + block_len <= len(data):
comment_parts.append(data[idx:idx+block_len].decode("utf-8", errors="ignore"))
idx += block_len
else:
break
comments.append("".join(comment_parts))
else:
while idx < len(data):
block_len = data[idx]
idx += 1
if block_len == 0:
break
idx += block_len
elif intro == 0x2c: # Image Descriptor
if idx + 10 > len(data):
break
packed = data[idx+9]
local_table = bool(packed & 0x80)
local_table_size = 2 ** ((packed & 0x07) + 1)
idx += 10
if local_table:
idx += 3 * local_table_size
idx += 1
while idx < len(data):
block_len = data[idx]
idx += 1
if block_len == 0:
break
idx += block_len
elif intro == 0x3b: # Trailer
break
else:
idx += 1
return "\n".join(comments)
# }}}