coming allong nicely. adding more commands, tests, and

artifacts for testing. basic framework down just adding features now
This commit is contained in:
venus
2026-07-18 02:38:13 -05:00
parent 8274d08f3e
commit 4dc0a24152
15 changed files with 362 additions and 209 deletions

79
src/ctf/cli_forensics.py Normal file
View File

@@ -0,0 +1,79 @@
# src/ctf/cli_forensics.py
# CLI wrapper and rendering layer for forensics commands
# vim foldmethod=marker
import click
from pathlib import Path
from ctf.forensics import get_metadata, COMMON_SIGNATURES
# {{{ forensics_group
@click.group(name="forensics")
def forensics_group():
""" A collection of forensics tools """
pass
# }}}
# {{{ tf
@forensics_group.command()
def tf():
click.echo("hello from forensics")
# }}}
# {{{ inspect
@forensics_group.command()
@click.argument('path', type=click.Path(exists=True))
def inspect(path):
"""Lists all basic inode metadata about a file"""
meta = get_metadata(Path(path))
# Present using Rich Table
from rich.console import Console
from rich.table import Table
console = Console()
table = Table(title=f"Metadata: {meta.filename}", show_header=False)
table.add_column("Key", style="bold cyan")
table.add_column("Value")
table.add_row("Filename", meta.filename)
table.add_row("Size", f"{meta.size} bytes")
table.add_row("Magic Bytes (Hex)", meta.magic)
table.add_row("Detected Type", meta.detected_type)
table.add_row("Extension", meta.extension)
table.add_row("Permissions", f"{meta.permissions_symbolic} ({meta.permissions_octal})")
table.add_row("Owner", f"{meta.owner_username} (UID: {meta.owner_uid})")
table.add_row("Group", f"{meta.owner_groupname} (GID: {meta.owner_gid})")
table.add_row("Allocated Space", f"{meta.allocated_size} bytes")
table.add_row("Hard Links", str(meta.hard_links))
table.add_row("Inode", str(meta.inode))
table.add_row("Device", str(meta.device))
if meta.extended_attributes:
xattr_str = ", ".join(f"{k}={v}" for k, v in meta.extended_attributes.items())
table.add_row("Extended Attributes", xattr_str)
console.print(table)
return meta
# }}}
# {{{ list_signatures
@forensics_group.command(name="signatures")
def list_signatures():
"""List all supported file magic signatures and expected extensions."""
from rich.console import Console
from rich.table import Table
console = Console()
table = Table(title="Supported Magic Signatures", show_header=True)
table.add_column("Magic Bytes (Hex)", style="bold cyan")
table.add_column("File Type", style="bold green")
table.add_column("Expected Exts")
for signature, (type_name, exts) in COMMON_SIGNATURES.items():
hex_str = signature.hex().upper()
exts_str = ", ".join(exts) if exts else "Any / None"
table.add_row(hex_str, type_name, exts_str)
console.print(table)
# }}}

View File

@@ -1,5 +1,7 @@
# functions for commands needed
# src/commands.py
# vim foldmethod=marker
import click
from pathlib import Path
@@ -48,11 +50,16 @@ def set_flag_format(pattern: str, original: str):
if "Competition" not in config:
config["Competition"] = {}
comp_name = config["Competition"].get("competition", "")
if original:
if not original.strip():
raise click.UsageError("Original flag cannot be empty or whitespace only.")
patterns = suggest_patterns(original)
click.echo("Suggested regex patterns:")
if comp_name and comp_name.lower() not in original.lower():
click.echo(f"Warning: Current competition name '{comp_name}' was not found in the example flag.")
patterns = suggest_patterns(original, comp_name=comp_name)
click.echo(f"Suggested regex patterns for competition {comp_name}:")
for i, pat in enumerate(patterns, 1):
click.echo(f" {i}. {pat}")
@@ -64,15 +71,22 @@ def set_flag_format(pattern: str, original: str):
click.echo(f"Flag format set to: {pattern}")
# }}}
# def Set_Challenge(comp: str, chal: str, setDirectory: bool):
# # set the current challenge and competition from input
# if state.current_comp != comp:
# state.current_comp=comp
# state.comp_dir=pathlib
# # TODO archive the old competitions
#
# if state.current_chal != chal:
# state.current_chal=chal
# print("challenge already set")
# # TODO archive the old challenges
# # TODO set the directory to challenge directory, with ignore option
# {{{ set_competition
# Sets the competition name in config.toml
@basic_group.command(name="set-competition")
@click.argument("name")
def set_competition(name: str):
"""Set the name of the active competition."""
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"]["competition"] = name
write_config(config, config_path)
click.echo(f"Competition name set to: {name}")
# }}}

View File

@@ -1,8 +1,7 @@
# src/ctf/forensics.py
# Library for forensic analysis
# Library for forensic analysis (pure functions only)
# vim foldmethod=marker
import click
from dataclasses import dataclass, field
from pathlib import Path
import stat
@@ -35,16 +34,6 @@ COMMON_SIGNATURES = {
b"ID3": ("MP3 Audio", [".mp3"]),
}
# {{{ forensics_group
@click.group(name="forensics")
def forensics_group():
''' A collection of forensics tools '''
pass
@forensics_group.command()
def tf(): print("hello from forensics")
# }}}
# {{{ FileMetadata
@dataclass
class FileMetadata:
@@ -78,12 +67,13 @@ class FileMetadata:
extended_attributes: Dict[str, str] = field(default_factory=dict)
# }}}
# {{{ inspect
@forensics_group.command()
@click.argument('path', type=click.Path(exists=True))
def inspect(path):
'''Lists all basic inode metadata about a file'''
# {{{ 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
@@ -156,7 +146,7 @@ def inspect(path):
except OSError:
pass
meta = FileMetadata(
return FileMetadata(
filename=p.name,
size=size,
magic=magic,
@@ -174,95 +164,5 @@ def inspect(path):
device=device,
extended_attributes=extended_attributes
)
# Present using Rich Table
from rich.console import Console
from rich.table import Table
console = Console()
table = Table(title=f"Metadata: {meta.filename}", show_header=False)
table.add_column("Key", style="bold cyan")
table.add_column("Value")
table.add_row("Filename", meta.filename)
table.add_row("Size", f"{meta.size} bytes")
table.add_row("Magic Bytes (Hex)", meta.magic)
table.add_row("Detected Type", meta.detected_type)
table.add_row("Extension", meta.extension)
table.add_row("Permissions", f"{meta.permissions_symbolic} ({meta.permissions_octal})")
table.add_row("Owner", f"{meta.owner_username} (UID: {meta.owner_uid})")
table.add_row("Group", f"{meta.owner_groupname} (GID: {meta.owner_gid})")
table.add_row("Allocated Space", f"{meta.allocated_size} bytes")
table.add_row("Hard Links", str(meta.hard_links))
table.add_row("Inode", str(meta.inode))
table.add_row("Device", str(meta.device))
if meta.extended_attributes:
xattr_str = ", ".join(f"{k}={v}" for k, v in meta.extended_attributes.items())
table.add_row("Extended Attributes", xattr_str)
console.print(table)
return meta
# }}}
# {{{ list_signatures
@forensics_group.command(name="signatures")
def list_signatures():
"""List all supported file magic signatures and expected extensions."""
from rich.console import Console
from rich.table import Table
console = Console()
table = Table(title="Supported Magic Signatures", show_header=True)
table.add_column("Magic Bytes (Hex)", style="bold cyan")
table.add_column("File Type", style="bold green")
table.add_column("Expected Exts")
for signature, (type_name, exts) in COMMON_SIGNATURES.items():
hex_str = signature.hex().upper()
exts_str = ", ".join(exts) if exts else "Any / None"
table.add_row(hex_str, type_name, exts_str)
console.print(table)
# }}}
# {{{ flag_detect
@forensics_group.command(name="flag-detect")
@click.argument("filepath", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--pattern", "-p", default=r"[a-zA-Z0-9_\-]+{[^}]+}", help="Regex pattern to search for.")
@click.option("--min-len", "-m", default=4, help="Minimum printable string length.")
def flag_detect(filepath: Path, pattern: str, min_len: int):
"""Extract printable strings and search for flag patterns."""
import re
from rich.console import Console
from rich.panel import Panel
console = Console()
try:
with open(filepath, "rb") as f:
content = f.read()
except Exception as e:
console.print(f"[bold red]Error reading file:[/bold red] {e}")
return
# Extract printable strings
printable_re = re.compile(rb"[a-zA-Z0-9/\-:.,_$%'\"()[\]<> ]{" + str(min_len).encode() + rb",}")
strings = printable_re.findall(content)
# Search for flags matching pattern
flag_re = re.compile(pattern.encode("utf-8"))
flags_found = []
for s in strings:
for match in flag_re.finditer(s):
flags_found.append(match.group().decode("utf-8", errors="ignore"))
if flags_found:
output = "\n".join(f" [bold green]✓[/bold green] {flag}" for flag in flags_found)
console.print(Panel(
output,
title=f"[bold green]Potential Flag(s) Found ({len(flags_found)})[/bold green]",
border_style="green"
))
else:
console.print("[bold yellow]No flag patterns found.[/bold yellow]")
# }}}

View File

@@ -2,8 +2,87 @@
# Parses and calls commands
from ctf.commands import basic_group
from ctf.forensics import forensics_group
from ctf.cli_forensics import forensics_group
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."""
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, nl=False)
else:
click.echo(f"Last detected flag: {last_flag}")
finally:
if _active_detector is not None:
_active_detector.detecting = True
# }}}
# {{{ main
def main():
@@ -12,7 +91,24 @@ def main():
cli.add_command(forensics_group)
cli.add_command(basic_group)
cli()
cli.add_command(flag_cmd)
from ctf.utils 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
try:
cli()
finally:
sys.stdout = detector.original_stream
_active_detector = None
else:
cli()
# }}}
if __name__ == "__main__":

View File

@@ -37,17 +37,14 @@ def active_competitions(dir: str) -> dict:
comps[item] = active_categories(item)
print(item.name)
print(comps[item])
return comps
# {{{ suggest_patterns
def suggest_patterns(s: str) -> list[str]:
"""Suggests a list of regex patterns from a sample flag string."""
def suggest_patterns(s: str, comp_name: str = "") -> list[str]:
"""Suggests a list of regex patterns from a sample flag string, optionally incorporating the competition name."""
if not s or not s.strip():
raise ValueError("Input string cannot be empty or whitespace only.")
# Group characters into types
groups = []
current_type = None
current_chars = []
def get_type(c):
if c.isupper():
return 'U'
@@ -57,19 +54,67 @@ def suggest_patterns(s: str) -> list[str]:
return 'D'
else:
return 'S'
for c in s:
ctype = get_type(c)
if ctype != current_type:
groups = []
comp_idx = -1
matched_prefix = ""
if comp_name and comp_name.strip():
comp_idx = s.lower().find(comp_name.lower())
if comp_idx != -1:
matched_prefix = s[comp_idx : comp_idx + len(comp_name)]
# Parse prefix before competition name
prefix_part = s[:comp_idx]
if prefix_part:
current_type = None
current_chars = []
for c in prefix_part:
ctype = get_type(c)
if ctype != current_type:
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
current_type = ctype
current_chars = [c]
else:
current_chars.append(c)
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
current_type = ctype
current_chars = [c]
else:
current_chars.append(c)
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
# Insert competition name group
groups.append(('C', matched_prefix))
# Parse suffix after competition name
suffix_part = s[comp_idx + len(comp_name):]
if suffix_part:
current_type = None
current_chars = []
for c in suffix_part:
ctype = get_type(c)
if ctype != current_type:
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
current_type = ctype
current_chars = [c]
else:
current_chars.append(c)
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
else:
# Normal grouping without competition name
current_type = None
current_chars = []
for c in s:
ctype = get_type(c)
if ctype != current_type:
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
current_type = ctype
current_chars = [c]
else:
current_chars.append(c)
if current_type is not None:
groups.append((current_type, "".join(current_chars)))
def escape_special(chars):
res = []
for c in chars:
@@ -79,55 +124,36 @@ def suggest_patterns(s: str) -> list[str]:
res.append(c)
return "".join(res)
# 1. Exact counts for group types
opt1_parts = []
for gtype, gchars in groups:
if gtype == 'U':
opt1_parts.append(f"[A-Z]{{{len(gchars)}}}")
elif gtype == 'L':
opt1_parts.append(f"[a-z]{{{len(gchars)}}}")
elif gtype == 'D':
opt1_parts.append(f"\\d{{{len(gchars)}}}")
else:
opt1_parts.append(escape_special(gchars))
opt1 = "^" + "".join(opt1_parts) + "$"
# 2. Variable counts for group types
opt2_parts = []
for gtype, gchars in groups:
if gtype == 'U':
opt2_parts.append("[A-Z]+")
elif gtype == 'L':
opt2_parts.append("[a-z]+")
elif gtype == 'D':
opt2_parts.append("\\d+")
else:
opt2_parts.append(escape_special(gchars))
opt2 = "^" + "".join(opt2_parts) + "$"
def build_pat(use_counts=True, merge_case=False):
parts = []
for gtype, gchars in groups:
if gtype == 'C':
parts.append(escape_special(gchars))
elif gtype == 'U':
if merge_case:
parts.append(f"[A-Za-z]{{{len(gchars)}}}" if use_counts else "[A-Za-z]+")
else:
parts.append(f"[A-Z]{{{len(gchars)}}}" if use_counts else "[A-Z]+")
elif gtype == 'L':
if merge_case:
parts.append(f"[A-Za-z]{{{len(gchars)}}}" if use_counts else "[A-Za-z]+")
else:
parts.append(f"[a-z]{{{len(gchars)}}}" if use_counts else "[a-z]+")
elif gtype == 'D':
parts.append(f"\\d{{{len(gchars)}}}" if use_counts else "\\d+")
else:
parts.append(escape_special(gchars))
return "".join(parts)
# 3. Case-insensitive / merged letters with exact counts
opt3_parts = []
for gtype, gchars in groups:
if gtype in ('U', 'L'):
opt3_parts.append(f"[A-Za-z]{{{len(gchars)}}}")
elif gtype == 'D':
opt3_parts.append(f"\\d{{{len(gchars)}}}")
else:
opt3_parts.append(escape_special(gchars))
opt3 = "^" + "".join(opt3_parts) + "$"
candidates = []
# 4. Case-insensitive / merged letters with variable counts
opt4_parts = []
for gtype, gchars in groups:
if gtype in ('U', 'L'):
opt4_parts.append("[A-Za-z]+")
elif gtype == 'D':
opt4_parts.append("\\d+")
else:
opt4_parts.append(escape_special(gchars))
opt4 = "^" + "".join(opt4_parts) + "$"
# 1. Unanchored patterns (ideal for searching/scanning)
candidates.append(build_pat(use_counts=True, merge_case=False))
candidates.append(build_pat(use_counts=False, merge_case=False))
candidates.append(build_pat(use_counts=True, merge_case=True))
candidates.append(build_pat(use_counts=False, merge_case=True))
# 5. General alphanumeric character class plus unique special characters
# Custom character class
unique_specials = set()
has_alpha = False
has_digit = False
@@ -139,7 +165,6 @@ def suggest_patterns(s: str) -> list[str]:
has_digit = True
else:
unique_specials.add(c)
char_class_parts = []
if has_alpha:
char_class_parts.append("A-Za-z")
@@ -150,13 +175,35 @@ def suggest_patterns(s: str) -> list[str]:
char_class_parts.append("\\" + spec)
else:
char_class_parts.append(spec)
opt5 = f"^[{''.join(char_class_parts)}]+$"
candidates.append(f"[{''.join(char_class_parts)}]+")
candidates.append(escape_special(s))
# 6. Literal exact escaped match
opt6 = "^" + escape_special(s) + "$"
# Wildcard patterns with .* and .*? (unanchored)
if comp_idx == 0:
suffix_part = s[len(comp_name):]
esc_prefix = escape_special(matched_prefix)
if suffix_part.startswith("{") and suffix_part.endswith("}"):
candidates.append(f"{esc_prefix}\\{{.*\\}}")
candidates.append(f"{esc_prefix}\\{{.*?\\}}")
candidates.append(f"{esc_prefix}\\{{[^}}]*\\}}")
candidates.append(f"{esc_prefix}\\{{[A-Za-z0-9_\\-]+\\}}")
inner_content = suffix_part[1:-1]
if all(c.islower() or c.isdigit() or c == '_' for c in inner_content):
candidates.append(f"{esc_prefix}\\{{[a-z0-9_]+\\}}")
else:
candidates.append(f"{esc_prefix}.*")
candidates.append(f"{esc_prefix}.*?")
else:
candidates.append(".*")
candidates.append(".*?")
# 2. Anchored versions of the above (ideal for strict validation)
anchored_candidates = []
for cand in candidates:
anchored_candidates.append(f"^{cand}$")
candidates.extend(anchored_candidates)
# Deduplicate while preserving order
candidates = [opt1, opt2, opt3, opt4, opt5, opt6]
seen = set()
result = []
for c in candidates: