From 75614f6a2182fd0f74b8b01b2591c3c34176112f Mon Sep 17 00:00:00 2001 From: venus Date: Sun, 19 Jul 2026 03:50:06 -0500 Subject: [PATCH] refactored the cli and forensics codebases as classes and into seperate folders for readability and consistent API access --- src/ctf/analysis.py | 197 ++++--- src/ctf/cli/__init__.py | 34 ++ src/ctf/{cli_analyse.py => cli/analyse.py} | 2 +- src/ctf/{commands.py => cli/basic.py} | 37 +- src/ctf/cli/flag.py | 47 ++ .../{cli_forensics.py => cli/forensics.py} | 8 +- src/ctf/{cli_steg.py => cli/steg.py} | 2 +- src/ctf/cli_helpers.py | 30 - src/ctf/config.py | 44 +- src/ctf/decoding.py | 208 ++++--- src/ctf/forensics/__init__.py | 32 ++ src/ctf/forensics/base.py | 114 ++++ src/ctf/forensics/gif.py | 98 ++++ src/ctf/{forensics.py => forensics/jpeg.py} | 544 +----------------- src/ctf/forensics/metadata.py | 139 +++++ src/ctf/forensics/png.py | 171 ++++++ src/ctf/helpers.py | 19 +- src/ctf/main.py | 26 +- src/ctf/utils.py | 2 +- 19 files changed, 991 insertions(+), 763 deletions(-) create mode 100644 src/ctf/cli/__init__.py rename src/ctf/{cli_analyse.py => cli/analyse.py} (98%) rename src/ctf/{commands.py => cli/basic.py} (77%) create mode 100644 src/ctf/cli/flag.py rename src/ctf/{cli_forensics.py => cli/forensics.py} (98%) rename src/ctf/{cli_steg.py => cli/steg.py} (99%) delete mode 100644 src/ctf/cli_helpers.py create mode 100644 src/ctf/forensics/__init__.py create mode 100644 src/ctf/forensics/base.py create mode 100644 src/ctf/forensics/gif.py rename src/ctf/{forensics.py => forensics/jpeg.py} (50%) create mode 100644 src/ctf/forensics/metadata.py create mode 100644 src/ctf/forensics/png.py diff --git a/src/ctf/analysis.py b/src/ctf/analysis.py index 20aaf87..8388dbc 100644 --- a/src/ctf/analysis.py +++ b/src/ctf/analysis.py @@ -1,6 +1,7 @@ # src/ctf/analysis.py # {{{ imports import re +from abc import ABC, abstractmethod from dataclasses import dataclass from pathlib import Path # }}} @@ -23,88 +24,132 @@ class AnalysisResult: routes: list[RouteResult] # }}} -# {{{ run_metadata_route -def run_metadata_route(file_path: Path) -> RouteResult: - """Extracts file metadata and scans for potential flags or hints.""" - from ctf.forensics import get_metadata - from ctf.decoding import check_for_flag - - file_path = Path(file_path).resolve() - if not file_path.exists(): - return RouteResult( - route_name="File Metadata Extraction", - executed=True, - success=False, - message=f"File not found: {file_path}" - ) +# {{{ BaseRoute +class BaseRoute(ABC): + """Abstract base class representing an analysis/solver route.""" + def __init__(self, route_name: str): + self.route_name = route_name + + @abstractmethod + def is_applicable(self, file_path: Path) -> bool: + """Determines if this route is applicable to the given file.""" + pass + + @abstractmethod + def execute(self, file_path: Path) -> RouteResult: + """Executes the analysis route on the target file.""" + pass +# }}} + +# {{{ MetadataRoute +class MetadataRoute(BaseRoute): + """Route that extracts metadata and parses comments/EXIF for flags.""" + def __init__(self): + super().__init__("File Metadata Extraction") + + def is_applicable(self, file_path: Path) -> bool: + # Metadata parsing is applicable to any file + return True + + def execute(self, file_path: Path) -> RouteResult: + from ctf.forensics import get_metadata + from ctf.decoding import check_for_flag - try: - meta = get_metadata(file_path) - extracted_flags = [] - - # Scan comment, EXIF attributes, and decoded hints - if meta.comment: - extracted_flags.extend(check_for_flag(meta.comment)) + file_path = Path(file_path).resolve() + if not file_path.exists(): + return RouteResult( + route_name=self.route_name, + executed=True, + success=False, + message=f"File not found: {file_path}" + ) - for val in meta.exif_data.values(): - if isinstance(val, str): - extracted_flags.extend(check_for_flag(val)) - - for dec_dict in meta.decoded_hints.values(): - for dec_val in dec_dict.values(): - extracted_flags.extend(check_for_flag(dec_val)) - - extracted_flags = list(set(extracted_flags)) - success = True - - if extracted_flags: - msg = f"Metadata inspected. Found {len(extracted_flags)} potential flag(s)." - else: - msg = "Metadata inspected. No obvious flags or decoded hints found." + try: + meta = get_metadata(file_path) + extracted_flags = [] - details = { - "Filename": meta.filename, - "Size": f"{meta.size} bytes", - "Detected Type": meta.detected_type, - "Magic": meta.magic, - "Has EXIF": "Yes" if meta.exif_data else "No", - "Has Comment": "Yes" if meta.comment else "No", - "Has Hints": "Yes" if meta.decoded_hints else "No", - } + # Scan comment, EXIF attributes, and decoded hints + if meta.comment: + extracted_flags.extend(check_for_flag(meta.comment)) + + for val in meta.exif_data.values(): + if isinstance(val, str): + extracted_flags.extend(check_for_flag(val)) + + for dec_dict in meta.decoded_hints.values(): + for dec_val in dec_dict.values(): + extracted_flags.extend(check_for_flag(dec_val)) + + extracted_flags = list(set(extracted_flags)) + success = True + + if extracted_flags: + msg = f"Metadata inspected. Found {len(extracted_flags)} potential flag(s)." + else: + msg = "Metadata inspected. No obvious flags or decoded hints found." + + details = { + "Filename": meta.filename, + "Size": f"{meta.size} bytes", + "Detected Type": meta.detected_type, + "Magic": meta.magic, + "Has EXIF": "Yes" if meta.exif_data else "No", + "Has Comment": "Yes" if meta.comment else "No", + "Has Hints": "Yes" if meta.decoded_hints else "No", + } + + return RouteResult( + route_name=self.route_name, + executed=True, + success=success, + message=msg, + details=details, + extracted_flags=extracted_flags + ) + except Exception as e: + return RouteResult( + route_name=self.route_name, + executed=True, + success=False, + message=f"Error extracting metadata: {str(e)}" + ) +# }}} + +# {{{ RoutingEngine +class RoutingEngine: + """Manages CTF challenge solvers and dynamic route execution.""" + def __init__(self): + self.routes: list[BaseRoute] = [] + + def register_route(self, route: BaseRoute): + self.routes.append(route) + + def run_all(self, file_path: Path | None) -> AnalysisResult: + executed_results = [] - return RouteResult( - route_name="File Metadata Extraction", - executed=True, - success=success, - message=msg, - details=details, - extracted_flags=extracted_flags - ) - except Exception as e: - return RouteResult( - route_name="File Metadata Extraction", - executed=True, - success=False, - message=f"Error extracting metadata: {str(e)}" - ) + if not file_path: + # Add a non-executed default result for the metadata route + executed_results.append(RouteResult( + route_name="File Metadata Extraction", + executed=False, + success=False, + message="No file attached to analyze." + )) + return AnalysisResult(success=False, routes=executed_results) + + resolved_path = Path(file_path).resolve() + for route in self.routes: + if route.is_applicable(resolved_path): + executed_results.append(route.execute(resolved_path)) + + success = any(r.success and r.extracted_flags for r in executed_results) + return AnalysisResult(success=success, routes=executed_results) # }}} # {{{ run_analysis def run_analysis(file_path: Path | None = None) -> AnalysisResult: - """Executes all registered analysis routes to analyze and solve a challenge.""" - routes = [] - - # Route 1: Metadata Extraction - if file_path: - routes.append(run_metadata_route(Path(file_path))) - else: - routes.append(RouteResult( - route_name="File Metadata Extraction", - executed=False, - success=False, - message="No file attached to analyze." - )) - - success = any(r.success and r.extracted_flags for r in routes) - return AnalysisResult(success=success, routes=routes) + """Compatibility function that instantiates the RoutingEngine and runs all routes.""" + engine = RoutingEngine() + engine.register_route(MetadataRoute()) + return engine.run_all(file_path) # }}} diff --git a/src/ctf/cli/__init__.py b/src/ctf/cli/__init__.py new file mode 100644 index 0000000..44edd5c --- /dev/null +++ b/src/ctf/cli/__init__.py @@ -0,0 +1,34 @@ +# src/ctf/cli/__init__.py +# {{{ imports +import click +import sys +from ctf.cli.basic import basic_group +from ctf.cli.forensics import forensics_group +from ctf.cli.steg import steg_group +from ctf.cli.analyse import analyse_cmd +from ctf.cli.flag import flag_cmd +# }}} + +# {{{ debug_exception_handler +def debug_exception_handler(type, value, tb): + """Unhandled exception hook to start post-mortem debugging in pdb.""" + import traceback + import pdb + traceback.print_exception(type, value, tb) + print("\n[!] Unhandled exception. Entering post-mortem debugger...") + pdb.post_mortem(tb) +# }}} + +# {{{ cli +@click.group() +@click.option("-d", "--debug", is_flag=True, help="Enable post-mortem debugging on errors.") +def cli(debug): + if debug: + sys.excepthook = debug_exception_handler + +cli.add_command(forensics_group) +cli.add_command(basic_group) +cli.add_command(flag_cmd) +cli.add_command(steg_group) +cli.add_command(analyse_cmd) +# }}} diff --git a/src/ctf/cli_analyse.py b/src/ctf/cli/analyse.py similarity index 98% rename from src/ctf/cli_analyse.py rename to src/ctf/cli/analyse.py index 3ac6f6a..fb11f49 100644 --- a/src/ctf/cli_analyse.py +++ b/src/ctf/cli/analyse.py @@ -1,4 +1,4 @@ -# src/ctf/cli_analyse.py +# src/ctf/cli/analyse.py # {{{ imports import click from pathlib import Path diff --git a/src/ctf/commands.py b/src/ctf/cli/basic.py similarity index 77% rename from src/ctf/commands.py rename to src/ctf/cli/basic.py index 1e9e116..0cb1292 100644 --- a/src/ctf/commands.py +++ b/src/ctf/cli/basic.py @@ -1,9 +1,8 @@ -# functions for commands needed -# src/commands.py - -# vim foldmethod=marker +# src/ctf/cli/basic.py +# {{{ imports import click from pathlib import Path +# }}} # {{{ basic_group # This defines a group with name basic which will nest other comands to be imported in the main loop @@ -37,7 +36,7 @@ def set_flag_format(pattern: str, original: str): You can either specify the regex PATTERN directly, or provide an example flag via the -o/--original option to generate and select from a list of suggested patterns. """ - from ctf.config import load_config, write_config + from ctf.config import Config from ctf.utils import suggest_patterns if pattern is None and original is None: @@ -45,13 +44,12 @@ def set_flag_format(pattern: str, original: str): if pattern is not None and original is not None: raise click.UsageError("Cannot specify both PATTERN and --original/-o option.") - config_path = "/home/venus/code/ctf/config.toml" - config = load_config(config_path) + cfg = Config() - if "Competition" not in config: - config["Competition"] = {} + if "Competition" not in cfg.data: + cfg.data["Competition"] = {} - comp_name = config["Competition"].get("competition", "") + comp_name = cfg.data["Competition"].get("competition", "") if original: if not original.strip(): @@ -67,8 +65,8 @@ def set_flag_format(pattern: str, original: str): val = click.prompt("Select a pattern index", type=click.IntRange(1, len(patterns))) pattern = patterns[val - 1] - config["Competition"]["flag_format"] = pattern - write_config(config, config_path) + cfg.data["Competition"]["flag_format"] = pattern + cfg.save(cfg.data) click.echo(f"Flag format set to: {pattern}") # }}} @@ -78,16 +76,13 @@ def set_flag_format(pattern: str, original: str): @click.argument("name") def set_competition(name: str): """Set the name of the active competition.""" - from ctf.config import load_config, write_config + from ctf.config import Config - config_path = "/home/venus/code/ctf/config.toml" - config = load_config(config_path) + cfg = Config() - if "Competition" not in config: - config["Competition"] = {} - - config["Competition"]["competition"] = name - write_config(config, config_path) + if "Competition" not in cfg.data: + cfg.data["Competition"] = {} + cfg.data["Competition"]["competition"] = name + cfg.save(cfg.data) click.echo(f"Competition name set to: {name}") # }}} - diff --git a/src/ctf/cli/flag.py b/src/ctf/cli/flag.py new file mode 100644 index 0000000..d831256 --- /dev/null +++ b/src/ctf/cli/flag.py @@ -0,0 +1,47 @@ +# src/ctf/cli/flag.py +# {{{ imports +import click +from ctf.helpers import detector_state +# }}} + +# {{{ flag_cmd +@click.command(name="flag") +@click.option("-p", "--plain", is_flag=True, help="Print raw flag without flavor text.") +@click.option("-l", "--list", "list_format", is_flag=True, help="List the current flag format.") +@click.option("-s", "--set", "set_format", type=str, help="Set the current flag format.") +@click.pass_context +def flag_cmd(ctx, plain, list_format, set_format): + """Retrieve the last detected flag from config, or list/set flag format options.""" + from ctf.config import Config + if list_format: + cfg = Config() + flag_format = cfg.data.get("Competition", {}).get("flag_format", "") + if plain: + click.echo(flag_format) + else: + click.echo(f"Current flag format: {flag_format}" if flag_format else "No flag format configured.") + return + + if set_format is not None: + from ctf.commands import set_flag_format + ctx.invoke(set_flag_format, pattern=set_format, original=None) + return + + # TODO add support for multiple flags and flag selection + if detector_state["active"] is not None: + detector_state["active"].detecting = False + try: + cfg = Config() + last_flag = cfg.data.get("Competition", {}).get("last_flag", "") + if not last_flag: + if not plain: + click.echo("No flag has been detected yet.") + return + if plain: + click.echo(last_flag) + else: + click.echo(f"Last detected flag: {last_flag}") + finally: + if detector_state["active"] is not None: + detector_state["active"].detecting = True +# }}} diff --git a/src/ctf/cli_forensics.py b/src/ctf/cli/forensics.py similarity index 98% rename from src/ctf/cli_forensics.py rename to src/ctf/cli/forensics.py index fc646bb..80f970a 100644 --- a/src/ctf/cli_forensics.py +++ b/src/ctf/cli/forensics.py @@ -1,10 +1,9 @@ -# src/ctf/cli_forensics.py -# CLI wrapper and rendering layer for forensics commands - -# vim foldmethod=marker +# src/ctf/cli/forensics.py +# {{{ imports import click from pathlib import Path from ctf.forensics import get_metadata, COMMON_SIGNATURES +# }}} # {{{ forensics_group @click.group(name="forensics") @@ -140,4 +139,3 @@ def list_signatures(): console.print(table) # }}} - diff --git a/src/ctf/cli_steg.py b/src/ctf/cli/steg.py similarity index 99% rename from src/ctf/cli_steg.py rename to src/ctf/cli/steg.py index b78c78b..54e13b0 100644 --- a/src/ctf/cli_steg.py +++ b/src/ctf/cli/steg.py @@ -1,4 +1,4 @@ -# src/ctf/cli_steg.py +# src/ctf/cli/steg.py # {{{ imports import click from pathlib import Path diff --git a/src/ctf/cli_helpers.py b/src/ctf/cli_helpers.py deleted file mode 100644 index 24174e3..0000000 --- a/src/ctf/cli_helpers.py +++ /dev/null @@ -1,30 +0,0 @@ -# src/ctf/cli_helpers.py -# {{{ imports -import click -from ctf.helpers import detector_state -# }}} - -# {{{ flag_cmd -@click.command(name="flag") -@click.option("-p", "--plain", is_flag=True, help="Print raw flag without flavor text.") -def flag_cmd(plain): - """Retrieve the last detected flag from config.""" - # TODO add support for multiple flags and flag selection - if detector_state["active"] is not None: - detector_state["active"].detecting = False - try: - from ctf.config import load_config - config = load_config("/home/venus/code/ctf/config.toml") - last_flag = config.get("Competition", {}).get("last_flag", "") - if not last_flag: - if not plain: - click.echo("No flag has been detected yet.") - return - if plain: - click.echo(last_flag) - else: - click.echo(f"Last detected flag: {last_flag}") - finally: - if detector_state["active"] is not None: - detector_state["active"].detecting = True -# }}} diff --git a/src/ctf/config.py b/src/ctf/config.py index cd8a858..4f4b804 100644 --- a/src/ctf/config.py +++ b/src/ctf/config.py @@ -1,26 +1,38 @@ # src/ctf/config.py # {{{ imports import toml +import os from pathlib import Path -from platformdirs import user_config_dir # }}} -# {{{ load_config -def load_config(config = f"{user_config_dir()}/ctf-config.toml") -> dict: - p = Path(config) - if p.exists(): - return toml.load(p) - return {} -# }}} - -# {{{ write_config -def write_config(data: dict, config = f"{user_config_dir()}/ctf"): - with open(config, "w") as f: - toml.dump(data, f) +# {{{ Config +class Config: + """Manages CTF challenge configuration loading and persistence.""" + def __init__(self, path: str | Path | None = None): + if path is None: + # Check environment variable first, then fallback to hardcoded path + path = os.environ.get("CTF_CONFIG_PATH", "/home/venus/code/ctf/config.toml") + self.path = Path(path) + self.data = self._load() + + def _load(self) -> dict: + if self.path.exists(): + try: + return toml.load(self.path) + except Exception: + return {} + return {} + + def save(self, data: dict): + self.data = data + self.path.parent.mkdir(parents=True, exist_ok=True) + with open(self.path, "w") as f: + toml.dump(self.data, f) # }}} # {{{ exports -config_data = load_config("/home/venus/code/ctf/config.toml") -competition = config_data.get("Competition", {}) -enviroment = config_data.get("Enviroment", {}) +# Load config instance to expose default values +_cfg = Config() +competition = _cfg.data.get("Competition", {}) +enviroment = _cfg.data.get("Enviroment", {}) # }}} diff --git a/src/ctf/decoding.py b/src/ctf/decoding.py index 34d4240..7156149 100644 --- a/src/ctf/decoding.py +++ b/src/ctf/decoding.py @@ -1,80 +1,152 @@ # src/ctf/decoding.py +# {{{ imports +from abc import ABC, abstractmethod from chepy import Chepy import re -from typing import Dict, Set +from typing import Dict, Set, List from ctf.helpers import check_for_flag, is_valid_flag # }}} -# {{{ attempt_decode -def attempt_decode(val: str, name: str, pattern: str, method_name: str, len_check=None, validator=None) -> str | None: - """Helper to run regex matching, dynamic Chepy decoding, and string printability checks.""" - # Find all matches sequentially and try decoding - for match in re.finditer(pattern, val): - matched_val = match.group(0) - if len_check and not len_check(matched_val): - continue - - try: - chepy_inst = Chepy(matched_val) - method = getattr(chepy_inst, method_name) - decoded = method().state - if isinstance(decoded, bytes): - dec_str = decoded.decode("utf-8") - elif isinstance(decoded, str): - dec_str = decoded - else: - dec_str = "" +# {{{ BaseDecoder +class BaseDecoder(ABC): + """Abstract base class representing a Chepy-based metadata decoder.""" + def __init__(self, name: str, pattern: str, method_name: str): + self.name = name + self.pattern = pattern + self.method_name = method_name + + def is_applicable(self, matched_val: str) -> bool: + """Determines if the matched substring is applicable for decoding (e.g. length checks).""" + return True + + def validate(self, decoded_str: str, original_str: str) -> bool: + """Validates if the decoded string is expected/correct (e.g. flag regex checks).""" + return True + + def attempt_single(self, val: str) -> str | None: + """Searches val for matches and returns the first successfully decoded printable string.""" + for match in re.finditer(self.pattern, val): + matched_val = match.group(0) + if not self.is_applicable(matched_val): + continue - if dec_str.strip() and all(32 <= ord(c) < 127 or c in "\r\n\t" for c in dec_str): - if validator and not validator(dec_str, val): - continue - return dec_str - except Exception: - pass - return None + try: + chepy_inst = Chepy(matched_val) + method = getattr(chepy_inst, self.method_name) + decoded = method().state + if isinstance(decoded, bytes): + dec_str = decoded.decode("utf-8") + elif isinstance(decoded, str): + dec_str = decoded + else: + dec_str = "" + + if dec_str.strip() and all(32 <= ord(c) < 127 or c in "\r\n\t" for c in dec_str): + if not self.validate(dec_str, val): + continue + return dec_str + except Exception: + pass + return None +# }}} + +# {{{ Concrete Decoders +class HexDecoder(BaseDecoder): + def __init__(self): + super().__init__("hex", r"[0-9a-fA-F]{4,}", "from_hex") + def is_applicable(self, matched_val: str) -> bool: + return len(matched_val) % 2 == 0 + +class Base64Decoder(BaseDecoder): + def __init__(self): + super().__init__("base64", r"[A-Za-z0-9+/=]{4,}", "from_base64") + def is_applicable(self, matched_val: str) -> bool: + return len(matched_val) % 4 == 0 + +class Base32Decoder(BaseDecoder): + def __init__(self): + super().__init__("base32", r"[A-Za-z2-7=]{4,}", "from_base32") + def is_applicable(self, matched_val: str) -> bool: + return len(matched_val) % 8 == 0 + +class BinaryDecoder(BaseDecoder): + def __init__(self): + super().__init__("binary", r"[01]{8,}", "from_binary") + def is_applicable(self, matched_val: str) -> bool: + return len(matched_val) % 8 == 0 + +class UrlDecoder(BaseDecoder): + def __init__(self): + super().__init__("url", r"(?:%[0-9a-fA-F]{2})+", "from_url_encoding") + +class Rot13Decoder(BaseDecoder): + def __init__(self): + super().__init__("rot13", r"[\x20-\x7E\s]{4,}", "rot_13") + def validate(self, decoded_str: str, original_str: str) -> bool: + return is_valid_flag(decoded_str, original_str) + +class ReversedDecoder(BaseDecoder): + def __init__(self): + super().__init__("reversed", r"[\x20-\x7E\s]{4,}", "reverse") + def validate(self, decoded_str: str, original_str: str) -> bool: + return is_valid_flag(decoded_str, original_str) +# }}} + +# {{{ DecodingRegistry +class DecodingRegistry: + """Orchestrates decoding metadata using registered BaseDecoder classes.""" + def __init__(self): + self.decoders: List[BaseDecoder] = [] + + def register_decoder(self, decoder: BaseDecoder): + self.decoders.append(decoder) + + def try_decode(self, val: str, current_chain: str = "", max_depth: int = 10, seen: Set[str] = None) -> Dict[str, str]: + if seen is None: + seen = set() + + results = {} + if not isinstance(val, str) or not val.strip() or max_depth <= 0: + return results + + cleaned_val = val.strip() + if cleaned_val in seen: + return results + seen.add(cleaned_val) + + for decoder in self.decoders: + decoded_val = decoder.attempt_single(cleaned_val) + if decoded_val is not None: + chain_name = f"{current_chain}->{decoder.name}" if current_chain else decoder.name + results[chain_name] = decoded_val + + # Recurse + nested_results = self.try_decode( + decoded_val, + current_chain=chain_name, + max_depth=max_depth - 1, + seen=set(seen) + ) + results.update(nested_results) + + return results + + @classmethod + def get_default_registry(cls) -> "DecodingRegistry": + registry = cls() + registry.register_decoder(HexDecoder()) + registry.register_decoder(Base64Decoder()) + registry.register_decoder(Base32Decoder()) + registry.register_decoder(BinaryDecoder()) + registry.register_decoder(UrlDecoder()) + registry.register_decoder(Rot13Decoder()) + registry.register_decoder(ReversedDecoder()) + return registry # }}} # {{{ try_decode_metadata def try_decode_metadata(val: str, current_chain: str = "", max_depth: int = 10, seen: Set[str] = None) -> Dict[str, str]: - """Attempts to decode a metadata value using defined formats recursively.""" - if seen is None: - seen = set() - - results = {} - if not isinstance(val, str) or not val.strip() or max_depth <= 0: - return results - - cleaned_val = val.strip() - if cleaned_val in seen: - return results - seen.add(cleaned_val) - - # We define our decoders here using unanchored patterns. Note: for rot13 and reversed, we only allow them - # if they produce a valid flag to avoid spamming everyday metadata fields. - formats = [ - ("hex", r"[0-9a-fA-F]{4,}", "from_hex", lambda s: len(s) % 2 == 0, None), - ("base64", r"[A-Za-z0-9+/=]{4,}", "from_base64", lambda s: len(s) % 4 == 0, None), - ("base32", r"[A-Za-z2-7=]{4,}", "from_base32", lambda s: len(s) % 8 == 0, None), - ("binary", r"[01]{8,}", "from_binary", lambda s: len(s) % 8 == 0, None), - ("url", r"(?:%[0-9a-fA-F]{2})+", "from_url_encoding", None, None), - ("rot13", r"[\x20-\x7E\s]{4,}", "rot_13", None, is_valid_flag), - ("reversed", r"[\x20-\x7E\s]{4,}", "reverse", None, is_valid_flag), - ] - - for name, pattern, method, len_check, validator in formats: - decoded_val = attempt_decode(cleaned_val, name, pattern, method, len_check, validator) - if decoded_val is not None: - chain_name = f"{current_chain}->{name}" if current_chain else name - results[chain_name] = decoded_val - - # Recurse with copy of seen to allow different branching paths to process same strings - nested_results = try_decode_metadata( - decoded_val, - current_chain=chain_name, - max_depth=max_depth - 1, - seen=set(seen) - ) - results.update(nested_results) - - return results + """Compatibility wrapper that routes decoding requests to the default registry.""" + registry = DecodingRegistry.get_default_registry() + return registry.try_decode(val, current_chain, max_depth, seen) # }}} diff --git a/src/ctf/forensics/__init__.py b/src/ctf/forensics/__init__.py new file mode 100644 index 0000000..fe5c7fd --- /dev/null +++ b/src/ctf/forensics/__init__.py @@ -0,0 +1,32 @@ +# src/ctf/forensics/__init__.py +# {{{ imports +from ctf.forensics.base import FileMetadata, FormatParser, ParserFactory, FallbackParser, COMMON_SIGNATURES +from ctf.forensics.jpeg import JpegParser +from ctf.forensics.png import PngParser +from ctf.forensics.gif import GifParser +from ctf.forensics.metadata import get_metadata +from pathlib import Path +from typing import Dict, Any +# }}} + +# {{{ get_exif +def get_exif(path: Path) -> Dict[str, Any]: + """Compatibility wrapper for extracting EXIF tags.""" + try: + with open(path, "rb") as f: + data = f.read() + return ParserFactory.get_parser(data).get_exif_tags(data) + except Exception: + return {} +# }}} + +# {{{ get_comment +def get_comment(path: Path) -> str: + """Compatibility wrapper for extracting comments.""" + try: + with open(path, "rb") as f: + data = f.read() + return ParserFactory.get_parser(data).get_comment(data) + except Exception: + return "" +# }}} diff --git a/src/ctf/forensics/base.py b/src/ctf/forensics/base.py new file mode 100644 index 0000000..4b0e086 --- /dev/null +++ b/src/ctf/forensics/base.py @@ -0,0 +1,114 @@ +# src/ctf/forensics/base.py +# {{{ imports +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Dict, Any +# }}} + +# {{{ 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 + + # POSIX Permissions + permissions_octal: str + permissions_symbolic: str + + # Ownership Identity + owner_uid: int + owner_username: str + owner_gid: int + owner_groupname: str + + # Allocation Metrics + allocated_size: int + + # Hard Links + hard_links: int + + # Inode & Device Identifiers + inode: int + device: int + + # 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 = "" +# }}} + +# {{{ FormatParser +class FormatParser(ABC): + """Abstract base class representing a file format metadata parser.""" + @abstractmethod + def is_matching(self, data: bytes) -> bool: + pass + @abstractmethod + def parse_physical(self, data: bytes) -> Dict[str, Any]: + pass + @abstractmethod + def get_comment(self, data: bytes) -> str: + pass + @abstractmethod + def get_exif_tags(self, data: bytes) -> Dict[str, Any]: + pass +# }}} + +# {{{ FallbackParser +class FallbackParser(FormatParser): + def is_matching(self, data: bytes) -> bool: + return True + def parse_physical(self, data: bytes) -> Dict[str, Any]: + return {} + def get_comment(self, data: bytes) -> str: + return "" + def get_exif_tags(self, data: bytes) -> Dict[str, Any]: + return {} +# }}} + +# {{{ ParserFactory +class ParserFactory: + @staticmethod + def get_parser(data: bytes) -> FormatParser: + from ctf.forensics.jpeg import JpegParser + from ctf.forensics.png import PngParser + from ctf.forensics.gif import GifParser + parsers = [JpegParser(), PngParser(), GifParser()] + for parser in parsers: + if parser.is_matching(data): + return parser + return FallbackParser() +# }}} diff --git a/src/ctf/forensics/gif.py b/src/ctf/forensics/gif.py new file mode 100644 index 0000000..59c4794 --- /dev/null +++ b/src/ctf/forensics/gif.py @@ -0,0 +1,98 @@ +# src/ctf/forensics/gif.py +# {{{ imports +import struct +from typing import Dict, Any +from ctf.forensics.base import FormatParser +# }}} + +# {{{ GifParser +class GifParser(FormatParser): + def is_matching(self, data: bytes) -> bool: + return data.startswith(b"GIF87a") or data.startswith(b"GIF89a") + def parse_physical(self, data: bytes) -> Dict[str, Any]: + return get_gif_physical(data) + def get_comment(self, data: bytes) -> str: + return extract_gif_comments(data) + def get_exif_tags(self, data: bytes) -> Dict[str, Any]: + return {} +# }}} + +# {{{ 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(" 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) +# }}} diff --git a/src/ctf/forensics.py b/src/ctf/forensics/jpeg.py similarity index 50% rename from src/ctf/forensics.py rename to src/ctf/forensics/jpeg.py index cb9cedb..843b696 100644 --- a/src/ctf/forensics.py +++ b/src/ctf/forensics/jpeg.py @@ -1,225 +1,35 @@ -# 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 +# src/ctf/forensics/jpeg.py +# {{{ imports 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"]), -} +from typing import Dict, Any +from ctf.forensics.base import FormatParser # }}} -# {{{ 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: +# {{{ JpegParser +class JpegParser(FormatParser): + def is_matching(self, data: bytes) -> bool: + return data.startswith(b"\xff\xd8\xff") + def parse_physical(self, data: bytes) -> Dict[str, Any]: + return get_jpeg_physical(data) + def get_comment(self, data: bytes) -> str: + return extract_jpeg_comment(data) + def get_exif_tags(self, data: bytes) -> Dict[str, Any]: + tags = {} + exif_data = extract_jpeg_exif(data) + if exif_data: + try: + tags.update(parse_tiff(exif_data)) + except Exception: pass + xmp_str = extract_jpeg_xmp(data) + if xmp_str: + try: + tags.update(parse_xmp(xmp_str)) + except Exception: pass 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 - ) + tags.update(extract_jpeg_iptc(data)) + except Exception: pass + return tags # }}} # {{{ EXIF tags @@ -260,7 +70,8 @@ EXIF_TAGS = { 0xa403: "WhiteBalance", 0xa405: "FocalLengthIn35mmFilm", 0xa406: "SceneCaptureType", -}# }}} +} +# }}} # {{{ parse_tiff def parse_tiff(data: bytes) -> Dict[str, Any]: @@ -371,7 +182,6 @@ def extract_jpeg_exif(data: bytes) -> bytes: 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 @@ -386,75 +196,6 @@ def extract_jpeg_exif(data: bytes) -> bytes: 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.""" @@ -479,23 +220,7 @@ def extract_jpeg_comment(data: bytes) -> str: 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 NS_MAP = { "http://www.w3.org/1999/02/22-rdf-syntax-ns#": "rdf", "http://creativecommons.org/ns#": "cc", @@ -504,7 +229,7 @@ NS_MAP = { "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]: @@ -654,84 +379,6 @@ def get_jpeg_physical(data: bytes) -> Dict[str, Any]: 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(" Dict[str, str]: """Extracts IPTC/NAA metadata (Record 2) from Photoshop APP13 segments.""" @@ -819,132 +466,3 @@ def extract_jpeg_iptc(data: bytes) -> Dict[str, str]: 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) -# }}} - diff --git a/src/ctf/forensics/metadata.py b/src/ctf/forensics/metadata.py new file mode 100644 index 0000000..e8230fa --- /dev/null +++ b/src/ctf/forensics/metadata.py @@ -0,0 +1,139 @@ +# src/ctf/forensics/metadata.py +# {{{ imports +from pathlib import Path +import os +import stat +from ctf.forensics.base import FileMetadata, COMMON_SIGNATURES, ParserFactory + +try: + import pwd + import grp +except ImportError: + pwd = None + grp = None +# }}} + +# {{{ 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() + size = stat_info.st_size + extension = p.suffix + + 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 + + mode = stat_info.st_mode + permissions_octal = oct(stat.S_IMODE(mode)) + permissions_symbolic = stat.filemode(mode) + + 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 + + if hasattr(stat_info, "st_blocks"): + allocated_size = stat_info.st_blocks * 512 + else: + allocated_size = size + + hard_links = stat_info.st_nlink + inode = stat_info.st_ino + device = stat_info.st_dev + + 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 + + try: + with open(p, "rb") as f: + file_bytes = f.read() + parser = ParserFactory.get_parser(file_bytes) + exif_data = parser.get_exif_tags(file_bytes) + comment = parser.get_comment(file_bytes) + physical_data = parser.parse_physical(file_bytes) + except Exception: + exif_data = {} + comment = "" + physical_data = {} + + 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 + ) +# }}} diff --git a/src/ctf/forensics/png.py b/src/ctf/forensics/png.py new file mode 100644 index 0000000..9de4c49 --- /dev/null +++ b/src/ctf/forensics/png.py @@ -0,0 +1,171 @@ +# src/ctf/forensics/png.py +# {{{ imports +import struct +from typing import Dict, Any +from ctf.forensics.base import FormatParser +# }}} + +# {{{ PngParser +class PngParser(FormatParser): + def is_matching(self, data: bytes) -> bool: + return data.startswith(b"\x89PNG\r\n\x1a\n") + def parse_physical(self, data: bytes) -> Dict[str, Any]: + return get_png_physical(data) + def get_comment(self, data: bytes) -> str: + return "" + def get_exif_tags(self, data: bytes) -> Dict[str, Any]: + tags = {} + exif_data = extract_png_exif(data) + if exif_data: + try: + from ctf.forensics.jpeg import parse_tiff + tags.update(parse_tiff(exif_data)) + except Exception: pass + try: + tags.update(parse_png_text_chunks(data)) + except Exception: pass + return tags +# }}} + +# {{{ 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_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 +# }}} + +# {{{ 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 +# }}} diff --git a/src/ctf/helpers.py b/src/ctf/helpers.py index 1a684c4..47ed7a0 100644 --- a/src/ctf/helpers.py +++ b/src/ctf/helpers.py @@ -18,9 +18,9 @@ def check_for_flag(text: str, pattern: str | None = None) -> list[str]: patterns.append(pattern.strip("^$")) else: try: - from ctf.config import load_config - config = load_config() - flag_format = config.get("Competition", {}).get("flag_format", "") + from ctf.config import Config + cfg = Config() + flag_format = cfg.data.get("Competition", {}).get("flag_format", "") except Exception: flag_format = "" if flag_format: @@ -74,13 +74,12 @@ class FlagDetectorStream: for match in matches: self.original_stream.write(f" \033[1;32m✓ {match}\033[0m\n") try: - from ctf.config import load_config, write_config - config_path = "/home/venus/code/ctf/config.toml" - config = load_config(config_path) - if "Competition" not in config: - config["Competition"] = {} - config["Competition"]["last_flag"] = match - write_config(config, config_path) + from ctf.config import Config + cfg = Config() + if "Competition" not in cfg.data: + cfg.data["Competition"] = {} + cfg.data["Competition"]["last_flag"] = match + cfg.save(cfg.data) except Exception: pass self._flag_buffer = "" diff --git a/src/ctf/main.py b/src/ctf/main.py index 1b5bbc7..61ab4c9 100644 --- a/src/ctf/main.py +++ b/src/ctf/main.py @@ -1,31 +1,15 @@ # src/main.py # Parses and calls commands -from ctf.commands import basic_group -from ctf.cli_forensics import forensics_group -from ctf.cli_steg import steg_group -from ctf.cli_analyse import analyse_cmd -from ctf.cli_helpers import flag_cmd -from ctf.helpers import FlagDetectorStream, detector_state -import click import sys - -# {{{ cli -@click.group() -def cli(): pass - -cli.add_command(forensics_group) -cli.add_command(basic_group) -cli.add_command(flag_cmd) -cli.add_command(steg_group) -cli.add_command(analyse_cmd) -# }}} +from ctf.cli import cli +from ctf.helpers import FlagDetectorStream, detector_state # {{{ main def main(): - from ctf.config import load_config - config = load_config("/home/venus/code/ctf/config.toml") - flag_format = config.get("Competition", {}).get("flag_format", "") + from ctf.config import Config + cfg = Config() + flag_format = cfg.data.get("Competition", {}).get("flag_format", "") if flag_format: detector = FlagDetectorStream(sys.stdout, flag_format) diff --git a/src/ctf/utils.py b/src/ctf/utils.py index e1c3d75..d44b088 100644 --- a/src/ctf/utils.py +++ b/src/ctf/utils.py @@ -2,7 +2,7 @@ # basic utilities from pathlib import Path -from ctf.config import load_config, competition, enviroment +from ctf.config import competition, enviroment # return a list of path objects for each catagory in a competition def active_categories(p: Path) -> list: