diff --git a/README.md b/README.md index 66e9a9c..a91a314 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ Pure-Python, zero-dependency parsers extract metadata without using heavy extern * `ctf flag`: Output the last captured flag from the configuration. ### Forensics Commands -* `ctf inspect `: Detailed metadata inspection. +* `ctf forensics metadata `: Detailed metadata inspection. * `-i, --inode`: Output POSIX file attributes only. * `-e, --exif`: Output EXIF/XMP/IPTC data and decoded hints. * `-p, --physical`: Output physical parameters of the target media. diff --git a/src/ctf/analysis.py b/src/ctf/analysis.py new file mode 100644 index 0000000..20aaf87 --- /dev/null +++ b/src/ctf/analysis.py @@ -0,0 +1,110 @@ +# src/ctf/analysis.py +# {{{ imports +import re +from dataclasses import dataclass +from pathlib import Path +# }}} + +# {{{ RouteResult +@dataclass +class RouteResult: + route_name: str + executed: bool + success: bool + message: str + details: dict | None = None + extracted_flags: list[str] | None = None +# }}} + +# {{{ AnalysisResult +@dataclass +class AnalysisResult: + success: bool + 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}" + ) + + 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)) + + 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="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)}" + ) +# }}} + +# {{{ 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) +# }}} diff --git a/src/ctf/cli_analyse.py b/src/ctf/cli_analyse.py new file mode 100644 index 0000000..3ac6f6a --- /dev/null +++ b/src/ctf/cli_analyse.py @@ -0,0 +1,55 @@ +# src/ctf/cli_analyse.py +# {{{ imports +import click +from pathlib import Path +from rich.console import Console +from rich.panel import Panel +from ctf.analysis import run_analysis +# }}} + +# {{{ analyse_cmd +@click.command(name="analyse") +@click.option('-f', '--file', type=click.Path(exists=True), help="Path to the challenge file to analyze.") +def analyse_cmd(file): + """Dynamically analyze and try to solve the active challenge using static routes.""" + console = Console() + console.print("[bold blue][*] Starting dynamic challenge analysis...[/bold blue]\n") + + file_path = Path(file) if file else None + result = run_analysis(file_path) + + for route in result.routes: + title = f"Route: {route.route_name}" + if not route.executed: + console.print(Panel( + f"[yellow]{route.message}[/yellow]", + title=title, + border_style="yellow" + )) + elif route.success: + content = f"[green]{route.message}[/green]\n" + if route.details: + content += "\n[bold cyan]Details:[/bold cyan]\n" + for k, v in route.details.items(): + content += f" • {k}: {v}\n" + if route.extracted_flags: + content += "\n[bold green]✓ Extracted Flags:[/bold green]\n" + for flag in route.extracted_flags: + content += f" [bold green]{flag}[/bold green]\n" + console.print(Panel( + content.strip(), + title=title, + border_style="green" + )) + else: + console.print(Panel( + f"[red]✗ {route.message}[/red]", + title=title, + border_style="red" + )) + + if result.success: + console.print("\n[bold green]✓ Analysis complete. Challenge solved successfully![/bold green]") + else: + console.print("\n[bold yellow]! Analysis complete. No clear flag found yet.[/bold yellow]") +# }}} diff --git a/src/ctf/cli_forensics.py b/src/ctf/cli_forensics.py index f96b733..fc646bb 100644 --- a/src/ctf/cli_forensics.py +++ b/src/ctf/cli_forensics.py @@ -13,13 +13,13 @@ def forensics_group(): pass # }}} -# {{{ inspect -@forensics_group.command() +# {{{ metadata +@forensics_group.command(name="metadata") @click.argument('path', type=click.Path(exists=True)) @click.option('-i', '--inode', is_flag=True, help="Print only POSIX inode metadata (without EXIF data).") @click.option('-e', '--exif', is_flag=True, help="Print only EXIF metadata (without POSIX inode metadata).") @click.option('-p', '--physical', is_flag=True, help="Print only physical image parameters.") -def inspect(path, inode, exif, physical): +def metadata(path, inode, exif, physical): """Lists all basic inode metadata and EXIF data about a file""" flags = [inode, exif, physical] if sum(flags) > 1: diff --git a/src/ctf/cli_helpers.py b/src/ctf/cli_helpers.py new file mode 100644 index 0000000..24174e3 --- /dev/null +++ b/src/ctf/cli_helpers.py @@ -0,0 +1,30 @@ +# 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/commands.py b/src/ctf/commands.py index d5d9ad1..1e9e116 100644 --- a/src/ctf/commands.py +++ b/src/ctf/commands.py @@ -37,7 +37,8 @@ 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.utils import load_config, write_config, suggest_patterns + from ctf.config import load_config, write_config + from ctf.utils import suggest_patterns if pattern is None and original is None: raise click.UsageError("Either PATTERN positional argument or --original/-o option must be specified.") @@ -77,7 +78,7 @@ 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.utils import load_config, write_config + from ctf.config import load_config, write_config config_path = "/home/venus/code/ctf/config.toml" config = load_config(config_path) diff --git a/src/ctf/config.py b/src/ctf/config.py new file mode 100644 index 0000000..cd8a858 --- /dev/null +++ b/src/ctf/config.py @@ -0,0 +1,26 @@ +# src/ctf/config.py +# {{{ imports +import toml +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) +# }}} + +# {{{ exports +config_data = load_config("/home/venus/code/ctf/config.toml") +competition = config_data.get("Competition", {}) +enviroment = config_data.get("Enviroment", {}) +# }}} diff --git a/src/ctf/decoding.py b/src/ctf/decoding.py index bffc599..34d4240 100644 --- a/src/ctf/decoding.py +++ b/src/ctf/decoding.py @@ -1,47 +1,8 @@ # src/ctf/decoding.py -# {{{ imports from chepy import Chepy import re from typing import Dict, Set -# }}} - -# {{{ is_valid_flag -def is_valid_flag(s: str, original: str = "") -> bool: - """ - Checks if the string contains a flag matching the active pattern or a generic pattern. - If original is provided, ensures that the decoded string contains a new unique flag - that was not already present in the original string. - """ - try: - from ctf.utils import load_config - config = load_config() - flag_format = config.get("Competition", {}).get("flag_format", "") - except Exception: - flag_format = "" - - patterns = [] - if flag_format: - patterns.append(flag_format.strip("^$")) - patterns.append(r"(?i)[a-z0-9_-]+{[a-z0-9_!@#$%^&*()\-+=]+}") - - decoded_flags = set() - for pattern in patterns: - for match in re.finditer(pattern, s): - decoded_flags.add(match.group(0)) - - if not decoded_flags: - return False - - if original: - original_flags = set() - for pattern in patterns: - for match in re.finditer(pattern, original): - original_flags.add(match.group(0)) - # If there are no new flags in decoded_flags, return False - if not (decoded_flags - original_flags): - return False - - return True +from ctf.helpers import check_for_flag, is_valid_flag # }}} # {{{ attempt_decode diff --git a/src/ctf/helpers.py b/src/ctf/helpers.py new file mode 100644 index 0000000..1a684c4 --- /dev/null +++ b/src/ctf/helpers.py @@ -0,0 +1,96 @@ +# {{{ imports +import re +# }}} + +detector_state = {"active": None} + + +# {{{ check_for_flag +def check_for_flag(text: str, pattern: str | None = None) -> list[str]: + """Scans the text for all flags matching the provided pattern, the active format, or a generic pattern. + Returns a list of matching flags found. + """ + if not text: + return [] + + patterns = [] + if pattern: + patterns.append(pattern.strip("^$")) + else: + try: + from ctf.config import load_config + config = load_config() + flag_format = config.get("Competition", {}).get("flag_format", "") + except Exception: + flag_format = "" + if flag_format: + patterns.append(flag_format.strip("^$")) + + patterns.append(r"(?i)[a-z0-9_-]+{[a-z0-9_!@#$%^&*()\-+=]+}") + + found = [] + for pat in patterns: + for match in re.finditer(pat, text): + found.append(match.group(0)) + + return list(set(found)) +# }}} + +# {{{ is_valid_flag +def is_valid_flag(s: str, original: str = "") -> bool: + """ + Checks if the string contains a flag matching the active pattern or a generic pattern. + If original is provided, ensures that the decoded string contains a new unique flag + that was not already present in the original string. + """ + flags = check_for_flag(s) + if not flags: + return False + if original: + original_flags = set(check_for_flag(original)) + # If there are no new flags in decoded flags, return False + if not (set(flags) - original_flags): + return False + return True +# }}} + +# {{{ FlagDetectorStream +class FlagDetectorStream: + """Intercepts stdout to scan output for matching flag formats.""" + def __init__(self, original_stream, pattern_str): + self.original_stream = original_stream + self._flag_buffer = "" + self.detecting = True + self.pattern_str = pattern_str + + def write(self, data): + self.original_stream.write(data) + if self.detecting and self.pattern_str: + self._flag_buffer += data + matches = check_for_flag(self._flag_buffer, self.pattern_str) + if matches: + self.detecting = False + self.original_stream.write("\n\033[1;32m[!] Potential flag(s) detected in command output:\033[0m\n") + 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) + except Exception: + pass + self._flag_buffer = "" + self.detecting = True + + def flush(self): + self.original_stream.flush() + + def __getattr__(self, name): + return getattr(self.original_stream, name) +# }}} + + diff --git a/src/ctf/main.py b/src/ctf/main.py index e38bfe4..1b5bbc7 100644 --- a/src/ctf/main.py +++ b/src/ctf/main.py @@ -2,89 +2,13 @@ # Parses and calls commands from ctf.commands import basic_group -from ctf.cli_forensics import forensics_group, inspect +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 -import re - -# Globally track the detector instance if it is active -_active_detector = None - -# {{{ FlagDetectorStream -class FlagDetectorStream: - """Intercepts stdout to scan output for matching flag formats.""" - def __init__(self, original_stream, pattern_str): - self.original_stream = original_stream - self._flag_buffer = "" - self.detecting = True - - # Clean pattern from standard anchors ^ and $ for substring matching - pat = pattern_str - if pat.startswith("^") and pat.endswith("$"): - pat = pat[1:-1] - try: - self.pattern = re.compile(pat) if pat else None - except re.error: - self.pattern = None - - def write(self, data): - self.original_stream.write(data) - if self.detecting and self.pattern: - self._flag_buffer += data - matches = [] - for match in self.pattern.finditer(self._flag_buffer): - matches.append(match.group()) - if matches: - self.detecting = False - self.original_stream.write("\n\033[1;32m[!] Potential flag(s) detected in command output:\033[0m\n") - for match in matches: - self.original_stream.write(f" \033[1;32m✓ {match}\033[0m\n") - try: - from ctf.utils 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) - except Exception: - pass - self._flag_buffer = "" - self.detecting = True - - def flush(self): - self.original_stream.flush() - - def __getattr__(self, name): - return getattr(self.original_stream, name) -# }}} - -# {{{ 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 - global _active_detector - if _active_detector is not None: - _active_detector.detecting = False - try: - from ctf.utils 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 _active_detector is not None: - _active_detector.detecting = True -# }}} # {{{ cli @click.group() @@ -93,26 +17,25 @@ def cli(): pass cli.add_command(forensics_group) cli.add_command(basic_group) cli.add_command(flag_cmd) -cli.add_command(inspect) cli.add_command(steg_group) +cli.add_command(analyse_cmd) # }}} # {{{ main def main(): - from ctf.utils import load_config + from ctf.config import load_config config = load_config("/home/venus/code/ctf/config.toml") flag_format = config.get("Competition", {}).get("flag_format", "") - global _active_detector if flag_format: detector = FlagDetectorStream(sys.stdout, flag_format) sys.stdout = detector - _active_detector = detector + detector_state["active"] = detector try: cli() finally: sys.stdout = detector.original_stream - _active_detector = None + detector_state["active"] = None else: cli() # }}} diff --git a/src/ctf/utils.py b/src/ctf/utils.py index 482e9fc..e1c3d75 100644 --- a/src/ctf/utils.py +++ b/src/ctf/utils.py @@ -1,23 +1,8 @@ # src/ctf/utils.py # basic utilities -import toml from pathlib import Path -from platformdirs import user_config_dir -# Parse the config file, returning a config dictionary with relevant config options - - -# Load the config from file and parse with TOML -def load_config(config = f"{user_config_dir()}/ctf-config.toml") -> dict: - p = Path(config) - if p.exists(): - return toml.load(p) - return{} - -# Write a dictionary to the config file -def write_config(data: dict, config = f"{user_config_dir()}/ctf"): - with open(config, "w") as f: - toml.dump(data, f) +from ctf.config import load_config, competition, enviroment # return a list of path objects for each catagory in a competition def active_categories(p: Path) -> list: @@ -213,10 +198,3 @@ def suggest_patterns(s: str, comp_name: str = "") -> list[str]: return result # }}} -# Load variables to export -config = load_config("/home/venus/code/ctf/config.toml") -competition = config["Competition"] -enviroment = config["Enviroment"] -# base_dir = config["ctf_dir"] - - diff --git a/tests/env/mock_analyse_flag.jpg b/tests/env/mock_analyse_flag.jpg new file mode 100644 index 0000000..ed1589d Binary files /dev/null and b/tests/env/mock_analyse_flag.jpg differ diff --git a/tests/env/mock_cli_analyse_flag.jpg b/tests/env/mock_cli_analyse_flag.jpg new file mode 100644 index 0000000..31ec3d7 Binary files /dev/null and b/tests/env/mock_cli_analyse_flag.jpg differ diff --git a/tests/env/test_write_config.toml b/tests/env/test_write_config.toml new file mode 100644 index 0000000..9e41f51 --- /dev/null +++ b/tests/env/test_write_config.toml @@ -0,0 +1 @@ +TestKey = "TestVal"