951 lines
33 KiB
Python
951 lines
33 KiB
Python
# src/ctf/forensics.py
|
|
# Library for forensic analysis (pure functions only)
|
|
|
|
# vim foldmethod=marker
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
import stat
|
|
import os
|
|
import sys
|
|
import struct
|
|
import xml.etree.ElementTree as ET
|
|
from typing import List, Dict, Any
|
|
|
|
try:
|
|
import pwd
|
|
import grp
|
|
except ImportError:
|
|
pwd = None
|
|
grp = None
|
|
|
|
# {{{ Common Signatures
|
|
COMMON_SIGNATURES = {
|
|
b"\x89PNG\r\n\x1a\n": ("PNG Image", [".png"]),
|
|
b"\xff\xd8\xff": ("JPEG Image", [".jpg", ".jpeg"]),
|
|
b"%PDF": ("PDF Document", [".pdf"]),
|
|
b"PK\x03\x04": ("ZIP Archive", [".zip"]),
|
|
b"\x7fELF": ("ELF Executable", [".elf"]),
|
|
b"MZ": ("PE Executable", [".exe", ".dll"]),
|
|
b"GIF87a": ("GIF Image", [".gif"]),
|
|
b"GIF89a": ("GIF Image", [".gif"]),
|
|
b"7z\xbc\xaf\x27\x1c": ("7-Zip Archive", [".7z"]),
|
|
b"\x1f\x8b": ("GZIP Archive", [".gz"]),
|
|
b"Rar!\x1a\x07\x00": ("RAR Archive", [".rar"]),
|
|
b"Rar!\x1a\x07\x01\x00": ("RAR Archive", [".rar"]),
|
|
b"BZh": ("BZIP2 Archive", [".bz2"]),
|
|
b"BM": ("BMP Image", [".bmp"]),
|
|
b"ID3": ("MP3 Audio", [".mp3"]),
|
|
}
|
|
# }}}
|
|
|
|
# {{{ FileMetadata
|
|
@dataclass
|
|
class FileMetadata:
|
|
filename: str
|
|
size: int
|
|
magic: str
|
|
extension: str
|
|
detected_type: str
|
|
|
|
# Task 1: POSIX Permissions
|
|
permissions_octal: str
|
|
permissions_symbolic: str
|
|
|
|
# Task 2: Ownership Identity
|
|
owner_uid: int
|
|
owner_username: str
|
|
owner_gid: int
|
|
owner_groupname: str
|
|
|
|
# Task 3: Allocation Metrics
|
|
allocated_size: int
|
|
|
|
# Task 4: Hard Links
|
|
hard_links: int
|
|
|
|
# Task 5: Inode & Device Identifiers
|
|
inode: int
|
|
device: int
|
|
|
|
# Task 6: Extended Attributes
|
|
extended_attributes: Dict[str, str] = field(default_factory=dict)
|
|
|
|
# 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 = ""
|
|
# }}}
|
|
|
|
# {{{ get_metadata
|
|
def get_metadata(path: Path) -> FileMetadata:
|
|
"""Extracts metadata attributes from a file without any console rendering."""
|
|
p = Path(path)
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"File not found: {p}")
|
|
|
|
stat_info = p.stat()
|
|
|
|
# Apparent size & extension
|
|
size = stat_info.st_size
|
|
extension = p.suffix
|
|
|
|
# Magic bytes
|
|
try:
|
|
with open(p, 'rb') as f:
|
|
magic = f.read(8).hex().upper()
|
|
except Exception:
|
|
magic = ""
|
|
|
|
detected_type = "Unknown"
|
|
try:
|
|
magic_bytes = bytes.fromhex(magic)
|
|
for signature, (type_name, exts) in COMMON_SIGNATURES.items():
|
|
if magic_bytes.startswith(signature):
|
|
detected_type = type_name
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# POSIX Permissions
|
|
mode = stat_info.st_mode
|
|
permissions_octal = oct(stat.S_IMODE(mode))
|
|
permissions_symbolic = stat.filemode(mode)
|
|
|
|
# Ownership Identity
|
|
owner_uid = stat_info.st_uid
|
|
owner_gid = stat_info.st_gid
|
|
owner_username = str(owner_uid)
|
|
owner_groupname = str(owner_gid)
|
|
|
|
if pwd is not None:
|
|
try:
|
|
owner_username = pwd.getpwuid(owner_uid).pw_name
|
|
except KeyError:
|
|
pass
|
|
if grp is not None:
|
|
try:
|
|
owner_groupname = grp.getgrgid(owner_gid).gr_name
|
|
except KeyError:
|
|
pass
|
|
|
|
# Allocation Metrics
|
|
if hasattr(stat_info, "st_blocks"):
|
|
allocated_size = stat_info.st_blocks * 512
|
|
else:
|
|
allocated_size = size
|
|
|
|
# Hard Links
|
|
hard_links = stat_info.st_nlink
|
|
|
|
# Inode & Device Identifiers
|
|
inode = stat_info.st_ino
|
|
device = stat_info.st_dev
|
|
|
|
# Extended Attributes
|
|
extended_attributes = {}
|
|
if hasattr(os, "listxattr") and hasattr(os, "getxattr"):
|
|
try:
|
|
attrs = os.listxattr(p)
|
|
for attr in attrs:
|
|
try:
|
|
val = os.getxattr(p, attr)
|
|
extended_attributes[attr] = val.decode("utf-8", errors="ignore")
|
|
except OSError:
|
|
pass
|
|
except OSError:
|
|
pass
|
|
|
|
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,
|
|
size=size,
|
|
magic=magic,
|
|
extension=extension,
|
|
detected_type=detected_type,
|
|
permissions_octal=permissions_octal,
|
|
permissions_symbolic=permissions_symbolic,
|
|
owner_uid=owner_uid,
|
|
owner_username=owner_username,
|
|
owner_gid=owner_gid,
|
|
owner_groupname=owner_groupname,
|
|
allocated_size=allocated_size,
|
|
hard_links=hard_links,
|
|
inode=inode,
|
|
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",
|
|
0x0110: "Model",
|
|
0x0112: "Orientation",
|
|
0x011a: "XResolution",
|
|
0x011b: "YResolution",
|
|
0x0128: "ResolutionUnit",
|
|
0x0131: "Software",
|
|
0x0132: "DateTime",
|
|
0x013b: "Artist",
|
|
0x8298: "Copyright",
|
|
0x829a: "ExposureTime",
|
|
0x829d: "FNumber",
|
|
0x8822: "ExposureProgram",
|
|
0x8825: "GPSInfo",
|
|
0x8827: "ISOSpeedRatings",
|
|
0x9000: "ExifVersion",
|
|
0x9003: "DateTimeOriginal",
|
|
0x9004: "DateTimeDigitized",
|
|
0x9201: "ShutterSpeedValue",
|
|
0x9202: "ApertureValue",
|
|
0x9204: "ExposureBiasValue",
|
|
0x9205: "MaxApertureValue",
|
|
0x9207: "MeteringMode",
|
|
0x9208: "LightSource",
|
|
0x9209: "Flash",
|
|
0x920a: "FocalLength",
|
|
0x9286: "UserComment",
|
|
0xa000: "FlashpixVersion",
|
|
0xa001: "ColorSpace",
|
|
0xa002: "ExifImageWidth",
|
|
0xa003: "ExifImageHeight",
|
|
0xa402: "ExposureMode",
|
|
0xa403: "WhiteBalance",
|
|
0xa405: "FocalLengthIn35mmFilm",
|
|
0xa406: "SceneCaptureType",
|
|
}# }}}
|
|
|
|
# {{{ parse_tiff
|
|
def parse_tiff(data: bytes) -> Dict[str, Any]:
|
|
"""Parses raw TIFF data block into EXIF tag dictionary."""
|
|
tags = {}
|
|
if len(data) < 8:
|
|
return tags
|
|
|
|
byte_order = data[0:2]
|
|
if byte_order == b"II":
|
|
endian = "<"
|
|
elif byte_order == b"MM":
|
|
endian = ">"
|
|
else:
|
|
return tags
|
|
|
|
magic = struct.unpack(f"{endian}H", data[2:4])[0]
|
|
if magic != 42:
|
|
return tags
|
|
|
|
ifd_offset = struct.unpack(f"{endian}I", data[4:8])[0]
|
|
|
|
try:
|
|
while ifd_offset != 0 and ifd_offset < len(data):
|
|
if ifd_offset + 2 > len(data):
|
|
break
|
|
num_entries = struct.unpack(f"{endian}H", data[ifd_offset:ifd_offset+2])[0]
|
|
entry_offset = ifd_offset + 2
|
|
|
|
for _ in range(num_entries):
|
|
if entry_offset + 12 > len(data):
|
|
break
|
|
tag, field_type, count, val_offset = struct.unpack(
|
|
f"{endian}HHII", data[entry_offset:entry_offset+12]
|
|
)
|
|
entry_offset += 12
|
|
|
|
# Resolve value based on type and count
|
|
val = None
|
|
type_sizes = {1: 1, 2: 1, 3: 2, 4: 4, 5: 8, 7: 1, 9: 4, 10: 8}
|
|
size = type_sizes.get(field_type, 1) * count
|
|
|
|
if size <= 4:
|
|
raw_val = struct.pack(f"{endian}I", val_offset)[:size]
|
|
else:
|
|
if val_offset + size <= len(data):
|
|
raw_val = data[val_offset:val_offset+size]
|
|
else:
|
|
raw_val = b""
|
|
|
|
if field_type == 2: # ASCII
|
|
val = raw_val.split(b"\x00")[0].decode("utf-8", errors="ignore")
|
|
elif field_type == 3: # SHORT
|
|
fmt = f"{endian}" + "H" * count
|
|
if len(raw_val) >= 2 * count:
|
|
val = struct.unpack(fmt, raw_val[:2*count])
|
|
if count == 1:
|
|
val = val[0]
|
|
elif field_type == 4: # LONG
|
|
fmt = f"{endian}" + "I" * count
|
|
if len(raw_val) >= 4 * count:
|
|
val = struct.unpack(fmt, raw_val[:4*count])
|
|
if count == 1:
|
|
val = val[0]
|
|
elif field_type == 5: # RATIONAL
|
|
fmt = f"{endian}" + "II" * count
|
|
if len(raw_val) >= 8 * count:
|
|
unpacked = struct.unpack(fmt, raw_val[:8*count])
|
|
rationals = [f"{unpacked[i*2]}/{unpacked[i*2+1]}" for i in range(count)]
|
|
val = rationals[0] if count == 1 else rationals
|
|
elif field_type == 7: # UNDEFINED
|
|
val = raw_val
|
|
elif field_type == 9: # SLONG
|
|
fmt = f"{endian}" + "i" * count
|
|
if len(raw_val) >= 4 * count:
|
|
val = struct.unpack(fmt, raw_val[:4*count])
|
|
if count == 1:
|
|
val = val[0]
|
|
elif field_type == 10: # SRATIONAL
|
|
fmt = f"{endian}" + "ii" * count
|
|
if len(raw_val) >= 8 * count:
|
|
unpacked = struct.unpack(fmt, raw_val[:8*count])
|
|
rationals = [f"{unpacked[i*2]}/{unpacked[i*2+1]}" for i in range(count)]
|
|
val = rationals[0] if count == 1 else rationals
|
|
else:
|
|
val = raw_val
|
|
|
|
tag_name = EXIF_TAGS.get(tag, f"Tag_{hex(tag)}")
|
|
tags[tag_name] = val
|
|
|
|
if entry_offset + 4 > len(data):
|
|
break
|
|
ifd_offset = struct.unpack(f"{endian}I", data[entry_offset:entry_offset+4])[0]
|
|
except Exception:
|
|
pass
|
|
|
|
return tags
|
|
# }}}
|
|
|
|
# {{{ extract_jpeg_exif
|
|
def extract_jpeg_exif(data: bytes) -> bytes:
|
|
"""Extracts raw Exif/TIFF payload from JPEG APP1 segments."""
|
|
if not data.startswith(b"\xff\xd8\xff"):
|
|
return b""
|
|
idx = 2
|
|
while idx < len(data) - 4:
|
|
if data[idx] == 0xff:
|
|
marker = data[idx+1]
|
|
if marker == 0xd9: # EOI
|
|
break
|
|
# Markers without length parameters
|
|
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 == 0xe1: # APP1
|
|
app1_data = data[idx+4:idx+4+length-2]
|
|
if app1_data.startswith(b"Exif\x00\x00"):
|
|
return app1_data[6:]
|
|
idx += 2 + length
|
|
else:
|
|
idx += 1
|
|
return b""
|
|
# }}}
|
|
|
|
# {{{ extract_png_exif
|
|
def extract_png_exif(data: bytes) -> bytes:
|
|
"""Extracts raw Exif/TIFF payload from PNG eXIf chunks."""
|
|
if not data.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
return b""
|
|
idx = 8
|
|
while idx < len(data) - 8:
|
|
length = struct.unpack(">I", data[idx:idx+4])[0]
|
|
chunk_type = data[idx+4:idx+8]
|
|
if chunk_type == b"eXIf":
|
|
return data[idx+8:idx+8+length]
|
|
elif chunk_type == b"IEND":
|
|
break
|
|
idx += 12 + length
|
|
return b""
|
|
# }}}
|
|
|
|
# {{{ get_exif
|
|
def get_exif(path: Path) -> Dict[str, Any]:
|
|
"""Reads file, extracts Exif and XMP segments, and parses them to a tag dictionary."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except Exception:
|
|
return {}
|
|
|
|
tags = {}
|
|
|
|
# 1. Parse standard EXIF
|
|
exif_data = b""
|
|
if data.startswith(b"\xff\xd8\xff"):
|
|
exif_data = extract_jpeg_exif(data)
|
|
elif data.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
exif_data = extract_png_exif(data)
|
|
|
|
if exif_data:
|
|
try:
|
|
tags.update(parse_tiff(exif_data))
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Parse XMP (Adobe)
|
|
xmp_str = ""
|
|
if data.startswith(b"\xff\xd8\xff"):
|
|
xmp_str = extract_jpeg_xmp(data)
|
|
|
|
if xmp_str:
|
|
try:
|
|
tags.update(parse_xmp(xmp_str))
|
|
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
|
|
# }}}
|
|
|
|
# {{{ extract_jpeg_comment
|
|
def extract_jpeg_comment(data: bytes) -> str:
|
|
"""Extracts raw comment string from JPEG COM (0xfe) segments."""
|
|
if not data.startswith(b"\xff\xd8\xff"):
|
|
return ""
|
|
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 == 0xfe: # COM
|
|
comment_data = data[idx+4:idx+4+length-2]
|
|
return comment_data.decode("utf-8", errors="ignore")
|
|
idx += 2 + length
|
|
else:
|
|
idx += 1
|
|
return ""
|
|
# }}}
|
|
|
|
# {{{ get_comment
|
|
def get_comment(path: Path) -> str:
|
|
"""Reads file, checks headers, and extracts JPEG COM / GIF comments."""
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except Exception:
|
|
return ""
|
|
|
|
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]:
|
|
"""Parses Adobe XMP XML string and returns namespace-prefixed tags."""
|
|
metadata = {}
|
|
try:
|
|
start_idx = xmp_str.find("<x:xmpmeta")
|
|
if start_idx == -1:
|
|
return metadata
|
|
end_idx = xmp_str.find("</x:xmpmeta>")
|
|
if end_idx == -1:
|
|
return metadata
|
|
xml_data = xmp_str[start_idx:end_idx + len("</x:xmpmeta>")]
|
|
|
|
root = ET.fromstring(xml_data)
|
|
|
|
def get_clean_name(tag: str) -> str:
|
|
if "}" in tag:
|
|
ns, name = tag.split("}")
|
|
ns = ns.lstrip("{")
|
|
prefix = NS_MAP.get(ns, ns.split("/")[-1].split("#")[0])
|
|
return f"{prefix}:{name}"
|
|
return tag
|
|
|
|
for elem in root.iter():
|
|
clean_tag = get_clean_name(elem.tag)
|
|
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
|
|
if elem.text and elem.text.strip():
|
|
val = elem.text.strip()
|
|
else:
|
|
attribs = {get_clean_name(k): v for k, v in elem.attrib.items()}
|
|
attribs.pop("rdf:about", None)
|
|
if "rdf:resource" in attribs:
|
|
val = attribs["rdf:resource"]
|
|
elif attribs:
|
|
val = ", ".join(f"{k} | {v}" if v else k for k, v in sorted(attribs.items()))
|
|
|
|
if val:
|
|
metadata[clean_tag] = val
|
|
except Exception:
|
|
pass
|
|
return metadata
|
|
# }}}
|
|
|
|
# {{{ extract_jpeg_xmp
|
|
def extract_jpeg_xmp(data: bytes) -> str:
|
|
"""Extracts XMP XML string from JPEG APP1 segments."""
|
|
if not data.startswith(b"\xff\xd8\xff"):
|
|
return ""
|
|
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 == 0xe1: # APP1
|
|
app1_data = data[idx+4:idx+4+length-2]
|
|
if app1_data.startswith(b"http://ns.adobe.com/xap/1.0/\x00"):
|
|
return app1_data[29:].decode("utf-8", errors="ignore")
|
|
idx += 2 + length
|
|
else:
|
|
idx += 1
|
|
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)
|
|
# }}}
|
|
|