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

cleaning things up by refactoring and combining files

Browse Source
This commit is contained in:
venus
2026-07-19 01:31:09 -05:00
parent 7f00bdb385
commit 185482f3d5
14 changed files with 335 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
src/ctf/analysis.py Normal file
View File

@@ -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)
# }}}

55
src/ctf/cli_analyse.py Normal file
View File

@@ -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]")
# }}}

6
src/ctf/cli_forensics.py
View File

@@ -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:

30
src/ctf/cli_helpers.py Normal file
View File

@@ -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
# }}}

5
src/ctf/commands.py
View File

@@ -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)

26
src/ctf/config.py Normal file
View File

@@ -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", {})
# }}}

41
src/ctf/decoding.py
View File

@@ -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

96
src/ctf/helpers.py Normal file
View File

@@ -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)
# }}}

93
src/ctf/main.py
View File

@@ -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()
# }}}

24
src/ctf/utils.py
View File

@@ -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"]