karl/collectionparser
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added pw cracking

Browse Source
This commit is contained in:
StasonJatham committed 2026-09-19 14:06:05 +02:00
1 parent 3205ab2311
commit b98f53e956
14 files changed
+606 -293
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+45 -7
View File
@@ -39,10 +39,12 @@ dedups each bucket. Peak disk is source + tree, so it fits where a global sort
would not.
```bash
# --scratch is a fast disk for per-batch files, sort temp, and build state.
# --batch_gb caps each batch (peak disk = source + tree).
collectionparser build \
--source /path/to/extracted \
--dataset /path/to/dataset \
--scratch /path/to/scratch \ # fast disk for batch files + sort temp
--scratch /path/to/scratch \
--batch_gb 120
```
@@ -50,14 +52,18 @@ collectionparser build \
same command. Finished batches are recorded in `build_state.json`, and parsed
files within the current batch in `progress_<n>.txt`; only unparsed files are
re-parsed. The sort+split step of an interrupted batch is redone (duplicates are
removed by the final dedup), so nothing is lost. Stats are written to
`--scratch/stats.md`.
removed by the final dedup), so nothing is lost.
State and stats live in `--scratch`: `build_state.json`, `stats.md`, and
`extract_failures.txt` (archives that failed to extract, never silently skipped).
`sort` temp is directed to `--scratch` via `TMPDIR`.
**Archives** (`.rar`/`.zip`/`.7z`/`.tar.*`/`.gz`/`.bz2`/`.xz`) are extracted
automatically into `--scratch` for the duration of their batch and removed once
parsed; the source is never modified. `.zip`/`.tar.*`/single-file `.gz`/`.bz2`/`.xz`
use the Python standard library; `.rar`/`.7z` need `7z` or `unrar` on `PATH` (use
`--no-extract` to skip archives entirely).
parsed; the source is never modified. `.zip`/`.tar.*`/single-file
`.gz`/`.bz2`/`.xz` use the Python standard library; `.rar`/`.7z` need `7z` or
`unrar` on `PATH`. Pass `--no-extract` to skip archives, or `--no-dedup` to
skip the final per-bucket de-duplication.
## Processing (manual)
@@ -90,6 +96,29 @@ collectionparser query --dataset data --pattern '*@gmail.com'
collectionparser export --dataset data --out records.jsonl
```
## Crack hashed passwords
`parse`/`build` keep hashed passwords (MD5/SHA1/SHA256) separate from plaintext
ones instead of discarding them. `build` collects them in `--scratch/hashes.txt`;
`parse` writes them to `--hashes_file` when given.
```bash
# Wordlist omitted -> rockyou is downloaded automatically; add the dataset's
# own plaintext passwords as extra candidates:
collectionparser crack \
--hashes hashes.txt \
--passwords-from dataset/ \
--out cracked.txt
# Or bring your own wordlist(s):
collectionparser crack --hashes hashes.txt --wordlist words.txt --out cracked.txt
```
Pure standard library: `hashlib` digests with one process per CPU core, each
attacking a strided slice of the wordlist, so memory stays bounded; candidate
sources are merged through `sort -u`. Only unsalted MD5/SHA1/SHA256 are
attackable — bcrypt is ignored. Feed `cracked.txt` back into the index with
`split`/`merge`.
## Measured performance
Single large-file test on Collections 2-5 (3 files, 19.94 GB, 3 workers):
@@ -114,7 +143,8 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
| `build` | Batched, resumable: parse + dedup + split into the query tree. |
| `query` | Exact or wildcard search over a dataset. |
| `export` | Export a dataset as JSON Lines. |
| `parse` | Parse an unstructured folder into `success`/`failure` files. |
| `crack` | Dictionary-attack collected hashes with a wordlist. |
| `parse` | Parse an unstructured folder into `success`/`failure`/`hashes` files. |
| `split` | Convert a large sorted file into a query-friendly `a/b/c` folder. |
| `chunk` | Split large TXT files into smaller ones. |
| `sort` | Sort and deduplicate every file in a query-friendly folder. |
@@ -123,3 +153,11 @@ Run `collectionparser --help` or `collectionparser <command> --help`.
| `test` | Look up passwords for the emails in a file. |
| `evaluate` | Measure hit rate of one dataset against another. |
| `filter-lines` | Keep lines whose length is within a range. |
## Development
```bash
pip install -e '.[dev]'
ruff check . # lint
pytest # tests
```
SKILL.md
+17 -12
View File
@@ -15,8 +15,8 @@ python -m venv venv && source venv/bin/activate
pip install -e /path/to/collectionparser
```
External tools it shells out to: `sort`, `split` (coreutils) and `rg`
(ripgrep, required only by `test`/`evaluate`).
External tools: `sort` and `split` (coreutils); `rg` (ripgrep) for
`query`/`test`/`evaluate`; optional `7z`/`unrar` to extract `.rar`/`.7z` archives.
## How to run it for the user
@@ -40,7 +40,6 @@ sort -T tmp -u -S 90% --parallel=12 success.txt > success_sorted.txt
collectionparser split --file success_sorted.txt --out data
# 4. Optional repair pass (moves records to the right bucket, re-sorts).
ulimit -n 60000
collectionparser clean --path data
# 5. Query.
@@ -58,9 +57,10 @@ collectionparser merge --src other_data --dest big_dataset
| Command | Purpose | Key options |
| --- | --- | --- |
| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb` |
| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb`, `--extract/--no-extract` |
| `query` | Exact or wildcard (`*`, `?`) search over a dataset. | `--dataset`, `--pattern` |
| `export` | Export a dataset as JSON Lines (OpenSearch). | `--dataset`, `--out` |
| `crack` | Dictionary-attack collected MD5/SHA1/SHA256 hashes. | `--hashes`, `--wordlist`, `--out`, `--workers` |
| `parse` | Folder of raw files -> `success`/`failure` files. Streams. | `--path`, `--success_file`, `--failure_file` |
| `split` | One sorted file -> `a/b/c` tree. | `--file`, `--out`, `--restart_from N` |
| `chunk` | Split oversized TXT files. | `--path`, `--size MB` (default 50) |
@@ -71,33 +71,38 @@ collectionparser merge --src other_data --dest big_dataset
| `evaluate` | Hit-rate of one index against another. | `--old`, `--new` |
| `filter-lines` | Keep lines within a length range. | `--file`, `--out`, `--less_than`, `--more_than` |
Global flag: `--debug` (verbose; also bypasses the open-files check below).
Global flag: `--debug` (verbose).
## Rules an agent must follow
- **Prefer `build` for large sources.** It parses in size-bounded batches so the
disk never needs source + temp + tree at once. Stop it anytime and rerun the
same command to resume: finished batches are in `--scratch/build_state.json`,
parsed files in `progress_<n>.txt`. Stats land in `--scratch/stats.md`.
parsed files in `progress_<n>.txt`. Stats land in `--scratch/stats.md`; any
archive that fails extraction is written to `--scratch/extract_failures.txt`.
- **Archives are auto-extracted** (`.zip`/`.tar.*`/`.gz` via stdlib; `.rar`/`.7z`
need `7z` or `unrar` on `PATH`). The source is never modified. Use
`--no-extract` to skip archive handling.
- **`split` requires a sorted, unique input.** Always run `sort -u` first; skip
it and the index silently contains duplicates and unsorted buckets.
- **`split` and `clean` need `ulimit -n >= 60000`.** The CLI refuses to run
otherwise. Raising the limit is the real fix; `--debug` only skips the check.
- **Memory is already flat.** Don't "optimize" by loading files in Python, and
don't pre-chunk unless a single file is unwieldy — the tools stream.
- **`merge` is destructive**: it appends into `--dest` and deletes each `--src`
file. Back up `--dest` before running.
- **`--dataset` is a path** to a query-friendly folder built by `split`.
- **`test`/`evaluate` need `rg` on `PATH`.** If `rg` is missing, queries return
nothing; install ripgrep instead of working around it.
- **`query`/`test`/`evaluate` need `rg` on `PATH`.** If `rg` is missing, queries
return nothing; install ripgrep instead of working around it.
- **`query` wildcards are anchored to the email field**: `*`/`?` never cross the
`:` separator (`*@gmail.com` works, matching into the password does not).
- **`parse` restarts safely.** Completed inputs are recorded in `progress.txt`
in the current directory; rerun the same command to resume. Don't delete it
mid-run.
- `parse` counts a `success` only for plaintext passwords. Hashes
(MD5/SHA1/SHA256/bcrypt) are intentionally not written to `success.txt`.
- `parse`/`build` write only plaintext passwords to `success`; hashed ones
(MD5/SHA1/SHA256) go to a separate hashes file (`--scratch/hashes.txt` from
`build`, `--hashes_file` from `parse`) so they can be attacked later:
`collectionparser crack --hashes hashes.txt --out cracked.txt` (omit
`--wordlist` to auto-download rockyou; add `--passwords-from dataset/` to reuse
known plaintext passwords). bcrypt is not attackable and is left untouched.
## Verify the environment
collectionparser/api.py
+78 -61
View File
@@ -1,3 +1,5 @@
"""Implementations behind the CLI commands: parse, split, sort, merge, query."""
import json
import os
import random
@@ -10,8 +12,7 @@ from tqdm import tqdm
from collectionparser.parser import parse_to_files
from collectionparser.query import QueryAPI
from collectionparser.splitter import get_output_file, write_line_to_file
from collectionparser.utils import (create_dir_for_file, filter_printable,
iter_lines, shuffle)
from collectionparser.utils import create_dir_for_file, filter_printable, iter_lines, shuffle
# ponytail: buckets under this size sort byte-exactly in RAM; larger buckets
# shell out to `sort -u`, which spills to disk.
@@ -19,16 +20,18 @@ _IN_MEMORY_SORT_LIMIT = 8 * 1024 * 1024
class API:
"""Command implementations shared by the CLI."""
@staticmethod
def evaluate(old, new):
def evaluate(old: str, new: str) -> None:
"""Report the hit rate of ``new`` against a sample of ``old``."""
hit_count = total_count = 0
files = [p for p in Path(old).expanduser().glob('**/*') if p.is_file()]
for _ in range(1000):
# reservoir sample 10 lines so arbitrarily large files stay O(1) in RAM.
random_lines = []
with open(random.choice(files), 'r', encoding='utf8', errors='ignore') as r:
for count, line in enumerate(r):
random_lines: list[str] = []
with open(random.choice(files), encoding='utf8', errors='ignore') as reader:
for count, line in enumerate(reader):
if count < 10:
random_lines.append(line.rstrip('\n'))
elif (j := random.randint(0, count)) < 10:
@@ -41,116 +44,127 @@ class API:
print(f'OLD: {random_line}, NEW: {hit}, HIT_RATE = {hit_count / total_count:.3f}.')
@staticmethod
def merge(src, dest):
def merge(src: str, dest: str) -> None:
"""Merge the bucket tree ``src`` into ``dest`` (destructive)."""
src_root = Path(src).expanduser()
dest_root = Path(dest).expanduser()
dest_files = {str(f).replace(str(dest_root), ''): f for f in find_files(dest_root)}
for src_file in find_files(src_root):
src_index = str(src_file).replace(str(src_root), '')
if src_index in dest_files:
merge(src_file, dest_files[src_index])
append_file(src_file, dest_files[src_index])
else:
dst = dest_root / src_index.lstrip('/')
create_dir_for_file(str(dst))
move(str(src_file), str(dst))
move_file(src_file, dst)
@staticmethod
def parse(path, success_file, failure_file):
parse_to_files(path, success_file, failure_file)
def parse(path: str, success_file: str, failure_file: str,
hashes_file: str | None = None) -> None:
parse_to_files(path, success_file, failure_file, hashes_file)
@staticmethod
def sort(path):
files = [p for p in Path(path).expanduser().glob('**/*') if p.is_file() and p.suffix != '.sorted']
for p in tqdm(files, desc='Unique > Sorting > Replacing existing files'):
output_path = p.with_suffix('.sorted')
sort_unique(p, output_path)
shutil.move(output_path, p)
def sort(path: str) -> None:
"""Sort and de-duplicate every file under ``path`` in place."""
files = [p for p in Path(path).expanduser().glob('**/*')
if p.is_file() and p.suffix != '.sorted']
for file in tqdm(files, desc='Unique > Sorting > Replacing existing files'):
output_path = file.with_suffix('.sorted')
sort_unique(file, output_path)
shutil.move(output_path, file)
@staticmethod
def clean(path):
def clean(path: str) -> None:
"""Move misplaced records to their correct bucket and sort each file."""
files = find_files(Path(path).expanduser())
shuffle(files)
num_files_moved = 0
files_handlers = {} # u limit here!
handles: dict[str, object] = {}
with tqdm(files, desc='clean') as bar:
for file in bar:
key = str(file).replace(path, '').removeprefix('/')
unique_file = str(file) + '.unique'
# stream: moved lines go to their target file, kept lines to a
# temp file. Nothing is held in RAM.
with open(str(file), 'r', encoding='utf8', errors='ignore') as r, \
open(unique_file, 'w', encoding='utf8') as w:
for line in r:
with open(str(file), encoding='utf8', errors='ignore') as reader, \
open(unique_file, 'w', encoding='utf8') as writer:
for line in reader:
email = line.partition(':')[0]
ground_truth_file = get_output_file(email)
if ground_truth_file != key:
output_file = Path(path) / ground_truth_file
write_line_to_file(files_handlers, line, output_file, buffer_scale=10)
if get_output_file(email) != key:
write_line_to_file(handles, line, Path(path) / get_output_file(email),
buffer_scale=10)
num_files_moved += 1
if num_files_moved % 100_000 == 0:
bar.set_description(f'clean. {num_files_moved:,} moved. '
f'move: {email}: {str(file).replace(path, "")}'
f' -> {str(output_file).replace(path, "")}')
bar.set_description(f'clean. {num_files_moved:,} moved.')
else:
w.write(line)
writer.write(line)
sort_unique(unique_file, file)
os.remove(unique_file)
@staticmethod
def test(file: str, dataset: str):
cracked_rate = total = 0
def test(file: str, dataset: str) -> None:
"""Look up every email in ``file`` and print the cracked rate."""
cracked = total = 0
for line in iter_lines(file):
results = QueryAPI.query(dataset, line.strip(), 'email')
if results:
passwords = [p.partition(':')[2] for p in results]
print('FOUND:', line, ':'.join(passwords))
cracked_rate += 1
cracked += 1
total += 1
print(f'Cracked rate: {cracked_rate / total if total else 0.0}')
print(f'Cracked rate: {cracked / total if total else 0.0}')
@staticmethod
def chunk_files(path: str, max_size: int):
# -d (numeric suffixes) is not compatible with MacOS.
def chunk_files(path: str, max_size: int) -> None:
"""Split files larger than ``max_size`` MB and delete the originals."""
files = find_files(Path(path).expanduser())
for f in tqdm(files, desc='chunk large files'):
if f.stat().st_size / 1e6 > max_size:
cmd = ['split', '-b', str(int(max_size * 1e6)), str(f), str(f) + '_']
print(f'File size: {f.stat().st_size / 1e6:.2f} Mb. ' + ' '.join(cmd))
for file in tqdm(files, desc='chunk large files'):
if file.stat().st_size / 1e6 > max_size:
# -d (numeric suffixes) is not compatible with MacOS.
cmd = ['split', '-b', str(int(max_size * 1e6)), str(file), str(file) + '_']
print(f'File size: {file.stat().st_size / 1e6:.2f} Mb. ' + ' '.join(cmd))
subprocess.run(cmd, check=True)
f.unlink()
file.unlink()
@staticmethod
def export_jsonl(dataset: str, out: str):
def export_jsonl(dataset: str, out: str) -> None:
"""Export the dataset as JSON Lines (`{"email":..., "password":...}`)."""
create_dir_for_file(out)
total = 0
with open(out, 'w', encoding='utf8') as w:
with open(out, 'w', encoding='utf8') as writer:
for path in tqdm(find_files(Path(dataset).expanduser()), desc='export jsonl'):
with open(path, 'r', encoding='utf8', errors='ignore') as r:
for line in r:
with open(path, encoding='utf8', errors='ignore') as reader:
for line in reader:
email, _, password = line.rstrip('\n').partition(':')
w.write(json.dumps({'email': email, 'password': password},
writer.write(json.dumps({'email': email, 'password': password},
ensure_ascii=False) + '\n')
total += 1
print(f'Exported {total:,} records to {out}.')
@staticmethod
def filter_lines(file, out, less_than, more_than):
def filter_lines(file: str, out: str, less_than: int, more_than: int) -> None:
"""Keep lines whose printable length is within ``[less_than, more_than]``."""
create_dir_for_file(out)
with open(out, 'w', encoding='utf8') as w, \
open(file, 'r', encoding='utf8', errors='ignore') as r:
for line in r:
line2 = filter_printable(line)
if less_than <= len(line2.strip()) <= more_than:
w.write(line2)
with open(out, 'w', encoding='utf8') as writer, \
open(file, encoding='utf8', errors='ignore') as reader:
for line in reader:
printable = filter_printable(line)
if less_than <= len(printable.strip()) <= more_than:
writer.write(printable)
def find_files(path):
def find_files(path) -> list[Path]:
"""Return every regular file under ``path`` (recursively)."""
root = Path(path).expanduser()
return [p for p in root.glob('**/*') if p.is_file()]
def sort_unique(src, dst):
def sort_unique(src, dst) -> None:
"""Write the sorted, de-duplicated lines of ``src`` to ``dst``.
In-memory for small files, ``sort -u`` for large ones. ``src`` may equal
``dst`` (in-place).
"""
src = Path(src)
if src.stat().st_size <= _IN_MEMORY_SORT_LIMIT:
lines = src.read_bytes().split(b'\n')
@@ -159,19 +173,22 @@ def sort_unique(src, dst):
lines = sorted(set(lines))
Path(dst).write_bytes(b'\n'.join(lines) + b'\n' if lines else b'')
else:
with open(dst, 'wb') as w:
subprocess.run(['sort', '-u', str(src)], stdout=w, check=True,
tmp = f'{dst}.tmp'
with open(tmp, 'wb') as writer:
subprocess.run(['sort', '-u', str(src)], stdout=writer, check=True,
env={**os.environ, 'LC_ALL': 'C'})
os.replace(tmp, dst)
def merge(src, dst):
def append_file(src, dst) -> None:
"""Append ``src`` to ``dst`` and delete ``src`` (streams, constant memory)."""
print(f'merge {src} {dst}')
with open(src, 'rb') as r, open(dst, 'ab') as w:
shutil.copyfileobj(r, w, length=1024 * 1024)
with open(src, 'rb') as reader, open(dst, 'ab') as writer:
shutil.copyfileobj(reader, writer, length=1024 * 1024)
print(f'del {src}')
os.remove(src)
def move(src, dst):
def move_file(src, dst) -> None:
print(f'mv {src} {dst}')
shutil.move(src, dst)
collectionparser/build.py
+56 -35
View File
@@ -27,7 +27,7 @@ import time
import zipfile
from pathlib import Path
from collectionparser.api import API
from collectionparser.api import API, find_files, sort_unique
from collectionparser.parser import parse_files
from collectionparser.splitter import split_file
@@ -41,7 +41,8 @@ def _is_archive(path: Path) -> bool:
return path.name.lower().endswith(ARCHIVE_SUFFIXES)
def _which(*names):
def _which(*names: str) -> str | None:
"""Return the path of the first available command among ``names``."""
for name in names:
found = shutil.which(name)
if found:
@@ -56,7 +57,7 @@ def _safe_target(dest: Path, name: str) -> Path:
return target
def _extract_compressed(path: Path, dest: Path):
def _extract_compressed(path: Path, dest: Path) -> None:
"""Try the available RAR/7z extractors in turn.
No single tool wins: `unrar` handles real RARs, `7z` handles the many
@@ -65,10 +66,7 @@ def _extract_compressed(path: Path, dest: Path):
"""
seven = _which('7z', '7za')
unrar = _which('unrar')
if path.name.lower().endswith('.rar'):
candidates = [unrar, seven]
else:
candidates = [seven, unrar]
candidates = [unrar, seven] if path.name.lower().endswith('.rar') else [seven, unrar]
last = None
for tool in (c for c in candidates if c):
shutil.rmtree(dest, ignore_errors=True)
@@ -77,14 +75,14 @@ def _extract_compressed(path: Path, dest: Path):
cmd = [tool, 'x', '-o+', '-idq', str(path), str(dest) + '/']
else:
cmd = [tool, 'x', '-y', f'-o{dest}', str(path)]
last = subprocess.run(cmd, stdout=DEVNULL, stderr=DEVNULL).returncode
last = subprocess.run(cmd, check=False, stdout=DEVNULL, stderr=DEVNULL).returncode
if last in (0, 1):
return
shutil.rmtree(dest, ignore_errors=True)
raise RuntimeError(f'extractors failed (last rc={last})')
def _extract_tar(path: Path, dest: Path):
def _extract_tar(path: Path, dest: Path) -> None:
with tarfile.open(path) as t:
try:
t.extractall(dest, filter='data') # py3.12+
@@ -92,7 +90,7 @@ def _extract_tar(path: Path, dest: Path):
t.extractall(dest)
def _extract_archive(path: Path, dest: Path):
def _extract_archive(path: Path, dest: Path) -> None:
name = path.name.lower()
dest.mkdir(parents=True, exist_ok=True)
if name.endswith('.zip'):
@@ -110,16 +108,16 @@ def _extract_archive(path: Path, dest: Path):
shutil.copyfileobj(r, w, length=1 << 20)
def _discover(source):
root = Path(source).expanduser()
files = [p for p in root.glob('**/*') if p.is_file()]
def _discover(source: str) -> list[Path]:
files = find_files(source)
files.sort(key=lambda p: p.stat().st_size) # small files first
print(f'{len(files)} files, {sum(1 for p in files if _is_archive(p))} archives.', flush=True)
return files
def _batches(files, batch_bytes):
batch, size = [], 0
def _batches(files: list[Path], batch_bytes: int):
batch: list[Path] = []
size = 0
for f in files:
sz = f.stat().st_size
if batch and size + sz > batch_bytes:
@@ -131,7 +129,7 @@ def _batches(files, batch_bytes):
yield batch
def _count_lines(path):
def _count_lines(path: str | Path) -> int:
total = 0
with open(path, 'rb') as r:
for chunk in iter(lambda: r.read(1 << 20), b''):
@@ -139,7 +137,7 @@ def _count_lines(path):
return total
def _stats_markdown(state):
def _stats_markdown(state: dict) -> str:
stats = state['stats']
total_in = sum(s['input_bytes'] for s in stats)
total_pairs = sum(s['pairs'] for s in stats)
@@ -156,6 +154,8 @@ def _stats_markdown(state):
]
if total_s:
lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s')
if state.get('hashes') is not None:
lines.append(f'- hashed passwords collected: {state["hashes"]:,}')
if state.get('dedup_s') is not None:
lines.append(f'- dedup seconds: {state["dedup_s"]:,.0f}')
if state.get('unique_records') is not None:
@@ -163,32 +163,42 @@ def _stats_markdown(state):
return '\n'.join(lines) + '\n'
def _extract_batch(archives, extract_dir, index):
def _extract_batch(archives: list[Path], extract_dir: Path,
index: int) -> tuple[list[Path], list[str]]:
"""Extract archives; return ``(files, failures)``.
Failures are logged, never silently dropped, so they can be recovered later.
"""
if not archives:
return []
return [], []
t0 = time.time()
failed = 0
errors = []
failures: list[str] = []
for i, archive in enumerate(archives):
try:
_extract_archive(archive, extract_dir / f'{i:05d}_{archive.name}')
except Exception as exc: # one bad archive must not stop the build
failed += 1
if len(errors) < 5:
errors.append(f'{archive.name}: {exc}')
failures.append(f'{archive}\t{exc}')
files = [p for p in extract_dir.glob('**/*') if p.is_file()]
print(f'batch {index}: {len(archives)} archives -> {len(files)} files, '
f'{failed} failed ({time.time() - t0:.0f}s)', flush=True)
if errors:
print(f'WARN: extract errors (first {len(errors)}): {errors}', flush=True)
return files
f'{len(failures)} failed ({time.time() - t0:.0f}s)', flush=True)
for failure in failures[:5]:
print(f'WARN: {failure}', flush=True)
return files, failures
def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True, extract=True):
def build(source: str, dataset: str, scratch: str, batch_bytes: int = DEFAULT_BATCH_BYTES,
dedup: bool = True, extract: bool = True) -> None:
"""Build the deduped query dataset from ``source`` into ``dataset``.
``scratch`` (a fast disk) holds per-batch files, sort temp and build state.
Safe to stop and rerun: completed batches are skipped and unparsed files of
the current batch are re-queued.
"""
dataset = Path(dataset).expanduser()
scratch = Path(scratch).expanduser()
dataset.mkdir(parents=True, exist_ok=True)
scratch.mkdir(parents=True, exist_ok=True)
os.environ['TMPDIR'] = str(scratch) # `sort -u` spills here, not into the dataset
for stale in scratch.glob('extract_*'):
shutil.rmtree(stale, ignore_errors=True)
@@ -204,6 +214,8 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True)
failures = scratch / 'failures.txt'
extract_failures = scratch / 'extract_failures.txt'
hashes = scratch / 'hashes.txt'
for index, batch in enumerate(batches):
if index in done:
continue
@@ -217,22 +229,27 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
progress_path.unlink(missing_ok=True)
parsed = set()
if success.exists() and progress_path.exists():
with open(progress_path, 'r', encoding='utf8', errors='ignore') as r:
with open(progress_path, encoding='utf8', errors='ignore') as r:
parsed = {line.rstrip('\n') for line in r}
inputs = [f for f in batch if not _is_archive(f)]
if extract:
inputs.extend(_extract_batch([f for f in batch if _is_archive(f)], extract_dir, index))
extracted, failures_ = _extract_batch(
[f for f in batch if _is_archive(f)], extract_dir, index)
inputs.extend(extracted)
if failures_:
with open(extract_failures, 'a', encoding='utf8') as w:
w.write('\n'.join(failures_) + '\n')
pending = [f for f in inputs if str(f) not in parsed]
t0 = time.time()
if pending:
parse_files(pending, str(success), str(failure), progress_filename=str(progress_path))
parse_files(pending, str(success), str(failure),
progress_filename=str(progress_path), hashes_filename=str(hashes))
if success.exists():
pairs = _count_lines(success)
rejects = _count_lines(failure) if failure.exists() else 0
subprocess.run(['sort', '-u', '-o', str(success), str(success)], check=True,
env={**os.environ, 'LC_ALL': 'C', 'TMPDIR': str(scratch)})
sort_unique(success, success)
split_file(str(success), str(dataset))
success.unlink(missing_ok=True)
if failure.exists():
@@ -255,8 +272,12 @@ def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True,
(scratch / 'stats.md').write_text(_stats_markdown(state))
print('BATCH ' + json.dumps(stat), flush=True)
if hashes.exists():
sort_unique(hashes, hashes) # de-duplicate collected hashed passwords
state['hashes'] = _count_lines(hashes)
state_path.write_text(json.dumps(state, indent=2))
if dedup and not state.get('deduped'):
os.environ['TMPDIR'] = str(scratch)
t0 = time.time()
API.sort(str(dataset)) # per-bucket sort -u dedups across batches
state['dedup_s'] = round(time.time() - t0, 1)
collectionparser/cli.py
+24 -13
View File
@@ -1,24 +1,18 @@
"""Command line entry point."""
import click
from collectionparser.api import API
from collectionparser.query import QueryAPI
from collectionparser.splitter import split_file
from collectionparser.utils import (input_dir, input_file, num_open_files_limit,
output_dir, output_file)
# Extraction: nohup find . -name '*.tar.gz' -execdir tar --keep-newer-files -xzvf '{}' -C ../extracted \; &
from collectionparser.utils import input_dir, input_file, output_dir, output_file
@click.group()
@click.option('--debug/--no-debug')
@click.pass_context
def cli(ctx, debug):
def cli(debug):
if debug:
click.echo('Debugging mode enabled.')
if not debug and num_open_files_limit() < 60_000 and ctx.invoked_subcommand in ('split', 'clean'):
click.echo('ERROR: Increase the limit of the number of open files to at least 60,000.')
click.echo('ERROR: ulimit -n 60000')
ctx.exit(1)
@cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). '
@@ -58,11 +52,13 @@ def test(file: str, dataset: str):
@cli.command(short_help='Parses an unstructured folder PATH of many files and generates two files: '
'SUCCESS_FILE and FAILURE_FILE. All valid email:password will go to SUCCESS_FILE.')
@click.option('--path', required=True, type=output_dir())
@click.option('--path', required=True, type=input_dir())
@click.option('--success_file', required=True, type=output_file())
@click.option('--failure_file', required=True, type=output_file())
def parse(path, success_file, failure_file):
API.parse(path, success_file, failure_file)
@click.option('--hashes_file', required=False, type=output_file(),
help='Collect hashed passwords here for later `crack`.')
def parse(path, success_file, failure_file, hashes_file):
API.parse(path, success_file, failure_file, hashes_file)
@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.')
@@ -109,6 +105,21 @@ def export(dataset: str, out: str):
API.export_jsonl(dataset, out)
@cli.command(short_help='Dictionary-attacks collected MD5/SHA1/SHA256 hashes.')
@click.option('--hashes', 'hashes_path', required=True, type=input_file())
@click.option('--wordlist', multiple=True, type=click.Path(),
help='Plain wordlist(s); missing files are downloaded from rockyou.')
@click.option('--passwords-from', multiple=True, type=click.Path(),
help='Dataset file/folder whose plaintext passwords join the wordlist.')
@click.option('--out', required=True, type=output_file())
@click.option('--workers', default=0, type=int, show_default=True,
help='0 = all CPUs.')
def crack(hashes_path: str, wordlist: tuple, passwords_from: tuple, out: str, workers: int):
from collectionparser.crack import crack as run_crack
run_crack(hashes_path, out, wordlists=wordlist, passwords_from=passwords_from,
workers=workers or None)
@cli.command(short_help='Builds the deduped query dataset from a raw source folder in batches. '
'Stop and rerun to resume.')
@click.option('--source', required=True, type=input_dir())
collectionparser/crack.py
+146
View File
@@ -0,0 +1,146 @@
"""Dictionary attacks over the unsalted hashes collected by ``parse``.
Candidate sources: plain wordlists, the password column of existing
``email:password`` datasets, and rockyou (downloaded on demand). They are merged
through ``sort -u`` so memory stays bounded for arbitrarily large inputs.
Only unsalted MD5/SHA1/SHA256 are attackable; bcrypt is intentionally not
supported (a wordlist attack on it is infeasible).
ponytail: the target index is rebuilt per worker (RAM x workers); a shared mmap
index is the upgrade path if the hash file is enormous.
"""
import gzip
import hashlib
import os
import shutil
import subprocess
import urllib.request
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from collectionparser.api import find_files
DEFAULT_WORDLIST_URL = 'https://weakpass.com/download/90/rockyou.txt.gz'
# hex digest length -> hashlib constructor
_ALGOS = {32: hashlib.md5, 40: hashlib.sha1, 64: hashlib.sha256}
_HEX_DIGITS = frozenset(b'0123456789abcdefABCDEF')
def download_rockyou(dest: Path) -> Path:
"""Download and gunzip the default rockyou wordlist if not already present."""
dest = Path(dest)
if dest.exists() and dest.stat().st_size:
return dest
dest.parent.mkdir(parents=True, exist_ok=True)
print(f'Downloading wordlist {DEFAULT_WORDLIST_URL} ...', flush=True)
with urllib.request.urlopen(DEFAULT_WORDLIST_URL) as response, \
gzip.GzipFile(fileobj=response) as gz, open(dest, 'wb') as out:
shutil.copyfileobj(gz, out, length=1 << 20)
return dest
def _write_passwords(source: Path, out) -> None:
"""Append the password column of an ``email:password`` file to ``out``."""
with open(source, encoding='utf8', errors='ignore') as reader:
for line in reader:
out.write(line.partition(':')[2].encode('utf8'))
def prepare_candidates(wordlists, passwords_from, work_dir: Path) -> Path:
"""Return a single, de-duplicated wordlist built from all candidate sources."""
work_dir = Path(work_dir)
work_dir.mkdir(parents=True, exist_ok=True)
wordlists = [Path(w) for w in wordlists] or [work_dir / 'rockyou.txt']
wordlists = [w if w.exists() else download_rockyou(w) for w in wordlists]
if len(wordlists) == 1 and not passwords_from:
return wordlists[0]
unique = work_dir / 'candidates.unique.txt'
env = {**os.environ, 'LC_ALL': 'C', 'TMPDIR': str(work_dir)}
process = subprocess.Popen(['sort', '-u'], stdin=subprocess.PIPE,
stdout=open(unique, 'wb'), env=env) # noqa: SIM115
with process.stdin as sink:
for wordlist in wordlists:
with open(wordlist, 'rb') as reader:
shutil.copyfileobj(reader, sink, length=1 << 20)
for source in passwords_from:
root = Path(source).expanduser()
for path in ([root] if root.is_file() else find_files(root)):
_write_passwords(path, sink)
process.wait()
if process.returncode != 0:
raise RuntimeError('failed to build the candidate wordlist')
return unique
def load_targets(hashes_path: str | Path) -> dict[int, dict[bytes, list[bytes]]]:
"""Return ``{hex_length: {digest_bytes: [email, ...]}}`` from a hashes file."""
targets: dict[int, dict[bytes, list[bytes]]] = {}
with open(hashes_path, 'rb') as reader:
for raw in reader:
line = raw.rstrip(b'\r\n')
email, separator, digest = line.rpartition(b':')
if not separator or len(digest) not in _ALGOS or not _HEX_DIGITS.issuperset(digest):
continue
try:
key = bytes.fromhex(digest.decode('ascii'))
except ValueError:
continue
targets.setdefault(len(digest), {}).setdefault(key, []).append(email)
return targets
def _scan_stripe(args: tuple) -> list[bytes]:
"""Attack one residue class of the wordlist (``index`` of every ``total``)."""
hashes_path, wordlist_path, index, total = args
targets = load_targets(hashes_path)
checks = [(targets[length], _ALGOS[length]) for length in targets]
matches: list[bytes] = []
with open(wordlist_path, 'rb') as reader:
for n, raw in enumerate(reader):
if n % total != index:
continue
word = raw.rstrip(b'\r\n')
if not word:
continue
for target, constructor in checks:
emails = target.get(constructor(word).digest())
if emails:
matches.extend(email + b':' + word + b'\n' for email in emails)
return matches
def crack(hashes_path: str, out_path: str, wordlists=(), passwords_from=(),
workers: int | None = None, work_dir: str | None = None) -> int:
"""Attack the hashes with the candidate wordlists; write ``email:password``."""
targets = load_targets(hashes_path)
if not targets:
print('No crackable MD5/SHA1/SHA256 hashes found.')
return 0
work_dir = Path(work_dir) if work_dir else Path(out_path).parent
wordlist = prepare_candidates(list(wordlists), list(passwords_from), work_dir)
workers = max(1, workers or os.cpu_count() or 1)
n_hashes = sum(len(emails) for by_digest in targets.values() for emails in by_digest.values())
print(f'Attacking {n_hashes:,} hashes with {wordlist.name} on {workers} worker(s).', flush=True)
jobs = [(str(hashes_path), str(wordlist), i, workers) for i in range(workers)]
total = 0
with open(out_path, 'wb') as writer:
if workers == 1:
results = map(_scan_stripe, jobs)
else:
executor = ProcessPoolExecutor(max_workers=workers)
results = executor.map(_scan_stripe, jobs)
try:
for matches in results:
for line in matches:
writer.write(line)
total += 1
finally:
if workers != 1:
executor.shutdown()
print(f'Cracked {total:,} hashes -> {out_path}.')
return total
collectionparser/parser.py
+60 -28
View File
@@ -4,6 +4,7 @@ import shutil
import string
import uuid
from concurrent.futures import ProcessPoolExecutor
from contextlib import ExitStack
from glob import glob
from pathlib import Path
from time import time
@@ -38,7 +39,8 @@ def password_type(password: str) -> PasswordType:
return PasswordType.DECRYPTED
def parse_line(row, separator, flip=False):
def parse_line(row: str, separator: str, flip: bool = False):
"""Try to read one ``email<sep>password`` record; ``None`` if implausible."""
if separator not in row:
return None
if separator != ' ':
@@ -50,7 +52,12 @@ def parse_line(row, separator, flip=False):
if flip:
mail, password = password, mail
password = password.replace(' ', '')
if not password or len(password) > 30 or '\n\n' in password:
if not password or '\n\n' in password:
return None
kind = password_type(password)
# Plaintext beyond 30 chars is junk; hashes are longer by nature and are kept
# so they can be attacked later with `crack`.
if kind == PasswordType.DECRYPTED and len(password) > 30:
return None
if set(password) == {'?'}: # [email protected]:?????? ??? ?????
return None
@@ -79,11 +86,12 @@ def parse_line(row, separator, flip=False):
if mail.count('@hotmail.com') == 2: # [email protected]@hotmail.com:8765432f
mail = mail.replace('@hotmail.com', '', 1)
if valid_email(mail) and not valid_email(password):
return mail, password, password_type(password)
return mail, password, kind
return None
def parseline(s: str):
"""Clean one raw line and return ``(email, password, type)`` or ``None``."""
b = filter_printable(s.strip())
while '..' in b:
b = b.replace('..', '')
@@ -113,9 +121,11 @@ class Processor:
per worker is a single read buffer regardless of file size.
"""
def __init__(self, success_filename=None, failure_filename=None):
def __init__(self, success_filename: str, failure_filename: str,
hashes_filename: str | None = None):
self.success_filename = success_filename
self.failure_filename = failure_filename
self.hashes_filename = hashes_filename
# Each worker writes its own buffered part file; the parent concatenates
# parts after every chunk. Buffered writes avoid the one-syscall-per-line
@@ -123,27 +133,39 @@ class Processor:
# interleaving (a >4KB line would otherwise split across writes).
# ponytail: parts cost one extra read+copy pass; merge in-worker only if
# that pass ever dominates.
def process(self, filename: Path):
def process(self, filename: Path) -> tuple[int, int]:
success = failure = 0
# Unique per task, not per process: one worker may pull several files
# from the queue, and a pid-named part would overwrite the previous one.
part = uuid.uuid4().hex
with open(str(filename), 'r', encoding='utf8', errors='ignore') as r, \
with open(str(filename), encoding='utf8', errors='ignore') as r, \
open(f'{self.success_filename}.part.{part}', 'w', encoding='utf8') as sw, \
open(f'{self.failure_filename}.part.{part}', 'w', encoding='utf8') as fw:
hashes = (open(f'{self.hashes_filename}.part.{part}', 'w', encoding='utf8') # noqa: SIM115
if self.hashes_filename else None)
try:
for line in r:
result = parseline(line)
if result is None:
fw.write(filter_printable(line)) # stop writing binary data to file.
fw.write(filter_printable(line)) # stop writing binary data.
failure += 1
elif result[2] == PasswordType.DECRYPTED:
mail, password, _ = result
continue
mail, password, kind = result
if kind == PasswordType.DECRYPTED:
sw.write(f'{str(mail).lower().strip()}:{password.strip()}\n')
success += 1
elif hashes is not None:
hashes.write(f'{str(mail).lower().strip()}:{password}\n')
else:
fw.write(filter_printable(line)) # no hash sink: keep as reject
failure += 1
finally:
if hashes is not None:
hashes.close()
return success, failure
def _clear_parts(*finals):
def _clear_parts(*finals: str) -> None:
# Leftovers mean a previous run died before merging. Those files were never
# recorded in progress.txt, so they are reprocessed; drop the partial parts.
for final in finals:
@@ -151,7 +173,7 @@ def _clear_parts(*finals):
os.remove(part)
def _merge_parts(final):
def _merge_parts(final: str) -> None:
parts = glob(f'{final}.part.*')
if not parts:
return
@@ -162,15 +184,24 @@ def _merge_parts(final):
os.remove(part)
def parse_files(files, success_filename, failure_filename, progress_filename=None):
"""Parse an explicit list of files; used by both `parse` and `build`."""
_clear_parts(success_filename, failure_filename)
def parse_files(files: list[Path], success_filename: str, failure_filename: str,
progress_filename: str | None = None,
hashes_filename: str | None = None) -> tuple[int, int]:
"""Parse an explicit list of files; used by both ``parse`` and ``build``.
Returns the total ``(success, failure)`` counts. Hashed passwords go to
``hashes_filename`` when given. When ``progress_filename`` is given, each
file is recorded after its chunk is safely merged, so a rerun can skip it.
"""
_clear_parts(success_filename, failure_filename,
*([hashes_filename] if hashes_filename else []))
success_total = failure_total = 0
workers = os.cpu_count() or 1
processor = Processor(success_filename, failure_filename)
processor = Processor(success_filename, failure_filename, hashes_filename)
start_time = time()
progress = open(progress_filename, 'a+', encoding='utf8') if progress_filename else None
try:
with ExitStack() as stack:
progress = (stack.enter_context(open(progress_filename, 'a', encoding='utf8'))
if progress_filename else None)
with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(files) as bar:
for i in range(0, len(files), workers):
chunk = files[i:i + workers]
@@ -181,25 +212,26 @@ def parse_files(files, success_filename, failure_filename, progress_filename=Non
# a crash never records a file whose output was not written.
_merge_parts(success_filename)
_merge_parts(failure_filename)
if hashes_filename:
_merge_parts(hashes_filename)
bar.update(len(chunk))
bar.set_description(f'success = {success_total:,} failure = {failure_total:,}')
if progress:
for path in chunk:
if progress:
progress.write(f'{path}\n')
finally:
if progress:
progress.close()
print(f'Time elapsed {time() - start_time:.3f} seconds.')
return success_total, failure_total
def parse_to_files(input_dir, success_filename, failure_filename):
# Restart from where we stopped. Stream the progress file; never build a
# second in-memory copy of it (it can grow to millions of lines).
done = set()
def parse_to_files(input_dir: str, success_filename: str, failure_filename: str,
hashes_filename: str | None = None) -> None:
"""Parse every file under ``input_dir``, resuming from ``progress.txt``."""
# Stream the progress file; never build a second in-memory copy of it (it
# can grow to millions of lines).
done: set[str] = set()
if os.path.exists('progress.txt'):
with open('progress.txt', 'r', encoding='utf8', errors='ignore') as p:
done = {line.rstrip('\n') for line in p}
with open('progress.txt', encoding='utf8', errors='ignore') as reader:
done = {line.rstrip('\n') for line in reader}
files = [p for p in Path(input_dir).glob('**/*') if p.is_file() and str(p) not in done]
shuffle(files)
parse_files(files, success_filename, failure_filename, 'progress.txt')
parse_files(files, success_filename, failure_filename, 'progress.txt', hashes_filename)
collectionparser/query.py
+39 -33
View File
@@ -1,12 +1,15 @@
"""Exact and wildcard queries over a bucketed dataset."""
import re
import subprocess
from functools import lru_cache
from pathlib import Path
def extract_emails_and_passwords(txt_lines):
emails_passwords = []
for txt_line in txt_lines:
def postprocess(results: list[str]) -> list[str]:
"""Normalise ``email:password`` lines into a sorted, de-duplicated list."""
emails_passwords: set[str] = set()
for txt_line in results:
if '@' not in txt_line: # does it contain an email address?
continue
if ':' in txt_line: # '[email protected]:iskandar89
@@ -19,46 +22,47 @@ def extract_emails_and_passwords(txt_lines):
email, password = txt_line.strip().split(separator)
except ValueError:
continue
emails_passwords.append((email, password))
return emails_passwords
def postprocess(results):
return sorted({':'.join(pair) for pair in extract_emails_and_passwords(results)})
emails_passwords.add(f'{email}:{password}')
return sorted(emails_passwords)
def glob_to_regex(pattern: str) -> str:
"""Glob -> regex anchored to the email field (`*`/`?` never cross the ':').
"""Convert a glob to a regex anchored to the email field.
ponytail: `*` matches within the email only; a `*` that must cross the
separator needs a different index.
``*`` and ``?`` never cross the ``:`` separator.
ponytail: ``*`` matches within the email only; a wildcard that must cross
the separator needs a different index.
"""
parts = []
for ch in pattern:
if ch == '*':
for char in pattern:
if char == '*':
parts.append('[^:]*')
elif ch == '?':
elif char == '?':
parts.append('[^:]')
else:
parts.append(re.escape(ch))
parts.append(re.escape(char))
return '^' + ''.join(parts) + ':'
class QueryAPI:
"""Read-only lookups: O(1) exact match or a ripgrep wildcard scan."""
@staticmethod
@lru_cache(maxsize=4096) # ponytail: cached dir listing is stale if the index changes mid-run
def scanning(directory: Path):
file_list, dir_list = {}, {}
for x in directory.iterdir():
if x.is_file():
file_list[x.name] = x
@lru_cache(maxsize=4096) # ponytail: stale if the index changes mid-run
def scanning(directory: Path) -> tuple[dict[str, Path], dict[str, Path]]:
"""Split a directory's entries into (files, subdirectories)."""
file_list: dict[str, Path] = {}
dir_list: dict[str, Path] = {}
for entry in directory.iterdir():
if entry.is_file():
file_list[entry.name] = entry
else:
dir_list[x.name] = x
dir_list[entry.name] = entry
return file_list, dir_list
@staticmethod
def search_in_file(path: Path, email):
def search_in_file(path: Path, email: str) -> list[str]:
"""Return matching lines for ``email`` from a single bucket file."""
assert path.is_file(), path
try:
output = subprocess.check_output(
@@ -68,21 +72,23 @@ class QueryAPI:
return [line for line in output.splitlines() if line]
@staticmethod
def recursive_search(path: Path, characters: list, index: int):
def recursive_search(path: Path, characters: list[str], index: int) -> list[str]:
"""Descend the bucket tree one character at a time, then search."""
if index >= len(characters):
return []
file_list, dir_list = QueryAPI.scanning(path)
cur = characters[index].lower()
if cur in file_list:
return QueryAPI.search_in_file(file_list[cur], ''.join(characters))
if cur in dir_list:
return QueryAPI.recursive_search(dir_list[cur], characters, index + 1)
current = characters[index].lower()
if current in file_list:
return QueryAPI.search_in_file(file_list[current], ''.join(characters))
if current in dir_list:
return QueryAPI.recursive_search(dir_list[current], characters, index + 1)
if 'symbols' in file_list:
return QueryAPI.search_in_file(file_list['symbols'], ''.join(characters))
return []
@staticmethod
def query(data_dir: str, request: str, query_type: str):
def query(data_dir: str, request: str, query_type: str) -> list[str]:
"""Exact email lookup using the bucket path (O(1) file read)."""
if len(request) <= 3:
return []
if query_type == 'email' and '@' not in request:
@@ -90,8 +96,8 @@ class QueryAPI:
return postprocess(QueryAPI.recursive_search(Path(data_dir), list(request), 0))
@staticmethod
def wildcard(data_dir: str, pattern: str):
"""Substring/wildcard search across the whole dataset with ripgrep.
def wildcard(data_dir: str, pattern: str) -> list[str]:
"""Glob search across the whole dataset with ripgrep.
ponytail: full scan of the tree (ripgrep is fast enough); add a
domain-sharded index only if wildcard latency becomes a problem.
collectionparser/splitter.py
+39 -23
View File
@@ -1,47 +1,63 @@
"""Turn a large ``email:password`` file into a query-friendly ``a/b/c`` tree."""
import io
import os
from functools import lru_cache
from itertools import takewhile
from time import time
from collectionparser.utils import RingDictionary, create_dir_for_file
from collectionparser.utils import create_dir_for_file
_ALNUM = 'abcdefghijklmnopqrstuvwxyz0123456789'
def split_file(breach_compilation_file, output_dir='out', restart_from=0):
# This can leave a few records in the wrong bucket; run `clean` afterwards.
# The cause is not fully understood, likely a limit on how many output files
# can stay open at once combined with buffering.
def split_file(input_file: str, output_dir: str = 'out', restart_from: int = 0) -> None:
"""Bucket every ``email:password`` line of ``input_file`` by email prefix.
Resume from the ``restart_from``-th input line. A few records may land in
the wrong bucket; run ``clean`` afterwards.
"""
os.makedirs(output_dir, exist_ok=True)
files_handlers = RingDictionary()
t1 = time()
with open(breach_compilation_file, 'r', encoding='utf8', errors='ignore') as r:
for i, line in enumerate(r):
handles: dict[str, object] = {}
tick = time()
with open(input_file, encoding='utf8', errors='ignore') as reader:
for i, line in enumerate(reader):
if i % 10_000_000 == 0:
elapsed = time() - t1
t1 = time()
elapsed = time() - tick
tick = time()
print(f'|- Processed {i:,} rows in {elapsed:.2f} seconds.')
if i < restart_from or ':' not in line:
continue
email = line.strip().split(':', 1)[0]
write_line_to_file(files_handlers, line, f'{output_dir}/{get_output_file(email)}')
write_line_to_file(handles, line, f'{output_dir}/{get_output_file(email)}')
def write_line_to_file(files_handlers, line, output_file, buffer_scale=100):
try: # faster than if: https://stackoverflow.com/questions/1835756/using-try-vs-if-in-python
files_handlers[output_file].write(line)
def write_line_to_file(handles: dict[str, object], line: str, output_file: str,
buffer_scale: int = 100, max_open: int = 100) -> None:
"""Append ``line`` to ``output_file``, keeping at most ``max_open`` handles.
Evicted handles are flushed and closed so the process stays under the file
descriptor limit.
"""
try:
handles[output_file].write(line) # type: ignore[attr-defined]
except KeyError:
if len(handles) >= max_open:
handles.pop(next(iter(handles))).flush() # type: ignore[attr-defined]
create_dir_for_file(output_file)
files_handlers[output_file] = open(output_file, mode='a+', encoding='utf8',
handles[output_file] = open( # noqa: SIM115 - cached handle, not scope-bound
output_file, mode='a+', encoding='utf8',
buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE)
files_handlers[output_file].write(line)
handles[output_file].write(line) # type: ignore[attr-defined]
@lru_cache(maxsize=8192)
def get_output_file(email: str) -> str:
"""Map an email to a b/c/d path using its first three alphanumeric chars."""
letters = list(takewhile(_ALNUM.__contains__, email.lower()[:3]))
if len(letters) < 3:
letters.append('symbols')
return '/'.join(letters)
"""Map an email to its bucket path from its first three alphanumeric chars."""
parts: list[str] = []
for char in email.lower()[:3]:
if char not in _ALNUM:
break
parts.append(char)
if len(parts) < 3:
parts.append('symbols')
return '/'.join(parts)
collectionparser/utils.py
+18 -43
View File
@@ -1,89 +1,64 @@
"""Small shared helpers: deterministic shuffling, line streaming, path handling."""
import os
import random
import re
import string
from enum import StrEnum
from resource import RLIMIT_NOFILE, getrlimit
import click
_NON_PRINTABLE = re.compile(f'[^{re.escape(string.printable)}]')
def shuffle(items):
random.seed(123)
random.shuffle(items)
def num_open_files_limit() -> int:
return getrlimit(RLIMIT_NOFILE)[0]
def shuffle(items: list) -> None:
"""Shuffle ``items`` in place with a fixed seed (reproducible ordering)."""
random.Random(123).shuffle(items)
def filter_printable(text: str) -> str:
"""Drop non-printable characters so no binary data is written to text files."""
return _NON_PRINTABLE.sub('', text)
def iter_lines(filename):
with open(filename, 'r', encoding='utf8', errors='ignore') as r:
for line in r:
def iter_lines(filename: str):
"""Yield lines from ``filename`` without the trailing newline, streaming."""
with open(filename, encoding='utf8', errors='ignore') as reader:
for line in reader:
yield line.rstrip('\n')
def create_dir_for_file(filename):
def create_dir_for_file(filename: str) -> None:
"""Create the parent directory of ``filename`` if it is missing."""
if parent := os.path.dirname(filename):
os.makedirs(parent, exist_ok=True)
def input_file():
def input_file() -> click.Path:
return click.Path(exists=True, file_okay=True, dir_okay=False,
writable=False, readable=True, resolve_path=True)
def input_dir():
def input_dir() -> click.Path:
return click.Path(exists=True, file_okay=False, dir_okay=True,
writable=False, readable=True, resolve_path=True)
def output_file():
def output_file() -> click.Path:
return click.Path(exists=False, file_okay=True, dir_okay=False,
writable=True, readable=True, resolve_path=True)
def output_dir():
def output_dir() -> click.Path:
return click.Path(exists=False, file_okay=False, dir_okay=True,
writable=True, readable=True, resolve_path=True)
class PasswordType(StrEnum):
"""Classification of a password field; only ``DECRYPTED`` is queryable."""
MD5 = 'MD5'
SHA1 = 'SHA1'
SHA256 = 'SHA256'
B_CRYPT = 'B_CRYPT'
DECRYPTED = 'DECRYPTED'
class RingDictionary:
"""Dict of at most ``max_num_keys`` entries; evicting flushes the oldest.
Used to keep a bounded number of open file handles while splitting a large
file into many output files.
"""
def __init__(self, max_num_keys=100):
self.d = {}
self.k = []
self.max_num_keys = max_num_keys
def __getitem__(self, key):
return self.d[key]
def __setitem__(self, key, item):
if key not in self.d:
if len(self.k) >= self.max_num_keys:
oldest = self.k.pop(0)
try:
self.d.pop(oldest).flush()
except Exception:
pass
self.k.append(key)
self.d[key] = item
pyproject.toml
+20 -1
View File
@@ -5,8 +5,18 @@ build-backend = "setuptools.build_meta"
[project]
name = "collectionparser"
version = "3.0.0"
description = "Command line tools to parse and query multi-billion password collections."
description = "Parse, deduplicate, and query multi-billion password collections."
readme = "README.md"
requires-python = ">=3.11"
keywords = ["passwords", "breach", "email", "dataset", "search"]
classifiers = [
"Environment :: Console",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.11",
"Topic :: Security",
"Topic :: Utilities",
]
dependencies = [
"click>=8.1",
"tqdm>=4.66",
@@ -14,9 +24,18 @@ dependencies = [
[project.optional-dependencies]
test = ["pytest>=8"]
dev = ["pytest>=8", "ruff>=0.6"]
[project.scripts]
collectionparser = "collectionparser.cli:cli"
[tool.setuptools.packages.find]
include = ["collectionparser*"]
[tool.ruff]
target-version = "py311"
line-length = 120
src = ["collectionparser", "tests"]
[tool.ruff.lint]
select = ["E4", "E7", "E9", "F", "I", "UP", "B", "SIM"]
tests/crack_test.py
+59
View File
@@ -0,0 +1,59 @@
import hashlib
import tempfile
import unittest
from glob import glob
from pathlib import Path
from collectionparser.crack import crack
from collectionparser.parser import Processor
class CrackTest(unittest.TestCase):
def test_hash_saved_by_processor(self):
with tempfile.TemporaryDirectory() as tmp:
src = Path(tmp) / 'in.txt'
src.write_text('[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n[email protected]:plainpw\n',
encoding='utf8')
processor = Processor(str(Path(tmp) / 's.txt'), str(Path(tmp) / 'f.txt'),
str(Path(tmp) / 'h.txt'))
processor.process(src)
hashes = ''.join(Path(p).read_text(encoding='utf8')
for p in glob(str(Path(tmp) / 'h.txt.part.*')))
success = ''.join(Path(p).read_text(encoding='utf8')
for p in glob(str(Path(tmp) / 's.txt.part.*')))
self.assertEqual('[email protected]:5f4dcc3b5aa765d61d8327deb882cf99\n', hashes)
self.assertEqual('[email protected]:plainpw\n', success)
def test_crack_md5_and_sha1(self):
with tempfile.TemporaryDirectory() as tmp:
hashes = Path(tmp) / 'hashes.txt'
hashes.write_bytes(
b'[email protected]:' + hashlib.md5(b'hunter2').hexdigest().encode() + b'\n'
b'[email protected]:' + hashlib.sha1(b'secret').hexdigest().encode() + b'\n'
b'[email protected]:not-a-hash\n')
wordlist = Path(tmp) / 'words.txt'
wordlist.write_bytes(b'foo\nhunter2\nbar\nsecret\n')
out = Path(tmp) / 'cracked.txt'
total = crack(str(hashes), str(out), wordlists=[str(wordlist)], workers=2)
self.assertEqual(2, total)
self.assertEqual({b'[email protected]:hunter2', b'[email protected]:secret'},
set(out.read_bytes().splitlines()))
def test_crack_from_plaintext_passwords(self):
with tempfile.TemporaryDirectory() as tmp:
hashes = Path(tmp) / 'hashes.txt'
hashes.write_bytes(b'[email protected]:' + hashlib.md5(b'reused').hexdigest().encode() + b'\n')
empty = Path(tmp) / 'empty.txt'
empty.write_bytes(b'')
plaintext = Path(tmp) / 'plain.txt'
plaintext.write_bytes(b'[email protected]:reused\n')
out = Path(tmp) / 'cracked.txt'
total = crack(str(hashes), str(out), wordlists=[str(empty)],
passwords_from=[str(plaintext)], workers=1)
self.assertEqual(1, total)
self.assertEqual([b'[email protected]:reused'], out.read_bytes().splitlines())
if __name__ == '__main__':
unittest.main()
tests/parser_test.py
+5 -1
View File
@@ -3,6 +3,7 @@ from time import time
from collectionparser.parser import parseline as p
from collectionparser.splitter import get_output_file
from collectionparser.utils import PasswordType
class ParserTest(unittest.TestCase):
@@ -97,7 +98,10 @@ class ParserTest(unittest.TestCase):
self.assertIsNone(p('300396 matelymaly [email protected] 222.155.173.78 '
'3d2c5414a041f518b03b750e62e4ebc3:Nhc'))
self.assertIsNone(p('05-02-2012;5721869;[email protected]:f02368945726d5fc2a14eb576f7276c0'))
# 32-char hex is a hash, not junk: kept (typed) so `crack` can attack it.
hashed = p('05-02-2012;5721869;[email protected]:f02368945726d5fc2a14eb576f7276c0')
self.assertIsNotNone(hashed)
self.assertEqual(PasswordType.MD5, hashed[2])
self.assertIsNotNone(p('[email protected]::g1u3ck'))
self.assertIsNotNone(p('[email protected]:[email protected]:2904198'))
self.assertIsNotNone(p('sanda [email protected]:vasafultv'))
tests/utils_test.py
-36
View File
@@ -1,36 +0,0 @@
import unittest
from collectionparser.utils import RingDictionary
class UtilsTest(unittest.TestCase):
def test_ring_dictionary_1(self):
rd = RingDictionary(max_num_keys=3)
rd['a'] = 2
self.assertListEqual(sorted(rd.d), ['a'])
self.assertListEqual(sorted(rd.k), ['a'])
rd['b'] = 2
self.assertListEqual(sorted(rd.d), ['a', 'b'])
self.assertListEqual(sorted(rd.k), ['a', 'b'])
rd['c'] = 2
self.assertListEqual(sorted(rd.d), ['a', 'b', 'c'])
self.assertListEqual(sorted(rd.k), ['a', 'b', 'c'])
rd['d'] = 2
self.assertListEqual(sorted(rd.d), ['b', 'c', 'd'])
self.assertListEqual(sorted(rd.k), ['b', 'c', 'd'])
rd = RingDictionary(max_num_keys=1)
rd['c'] = 1
rd['c'] = 2
rd['c'] = 3
self.assertListEqual(sorted(rd.d), ['c'])
self.assertListEqual(sorted(rd.k), ['c'])
rd['d'] = 3
self.assertListEqual(sorted(rd.d), ['d'])
self.assertListEqual(sorted(rd.k), ['d'])
if __name__ == '__main__':
unittest.main()