diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml deleted file mode 100644 index b3e55b3..0000000 --- a/.github/FUNDING.yml +++ /dev/null @@ -1 +0,0 @@ -github: [philipperemy] diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0236f61 --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.coverage +htmlcov/ +.tox/ +.nox/ +.venv/ +venv/ +build/ +dist/ +*.egg-info/ + +# Local datasets and pipeline scratch/state files +data/ +scratch/ +*.jsonl +build_state.json +progress_*.txt + +.DS_Store \ No newline at end of file diff --git a/Makefile b/Makefile deleted file mode 100644 index 75e2f6a..0000000 --- a/Makefile +++ /dev/null @@ -1,16 +0,0 @@ -all: - make install - -install: - pip install cython pytest - python setup.py build_ext --inplace - pip install . --upgrade - -uninstall: - pip uninstall -y breach - -test: - pytest tests - -clean: - rm -rf *.out breach/*.c *.bin *.exe *.o *.a breach/*.so build *.html __pycache__ breach/__pycache__ diff --git a/README.md b/README.md index c7c0d4a..ad6a5f8 100644 --- a/README.md +++ b/README.md @@ -1,177 +1,125 @@ -# Collections 1, Collections 2-5 and AntiPublic Parser +# collectionparser -Command line tools to manipulate the data from those multi-billion passwords collections. +Command line tools to parse, deduplicate, and query multi-billion password +collections (Collections 1-5, AntiPublic, Breach Compilation). -The full processing will take a couple of days and will generate a file structure that can be queried almost in o(1). +Processing produces a deduped `a/b/c` bucket tree that answers exact lookups in +O(1) and wildcard queries with a parallel ripgrep scan: ``` -$ john.applessed@apple.com -john.applessed@apple.com:toto123 +$ collectionparser query --dataset /path/to/data --pattern john.smith@gmail.com +john.smith@gmail.com:toto123 +$ collectionparser query --dataset /path/to/data --pattern '*@gmail.com' ``` -The total number of unique records in the final dataset (Collection 1 to 5 + AntiPublic + Breach Compilation) is around 3.72 billions (_3,372,591,561_ to be precise). +## Install - -### Set Up - -Create a virtual environment and install the package. - -``` -virtualenv -p python3 venv -source venv/bin/activate -make install -``` - -## Extraction - -https://superuser.com/questions/1308374/how-to-extract-all-tar-gz-files-present-in-multiple-folders-at-a-time-to-another +Requires Python 3.11+ and [`ripgrep`](https://github.com/BurntSushi/ripgrep) (`rg`) for queries. ```bash -find "$1" -name '*.tar.gz' -execdir tar -xzvf '{}' -C extracted \; -find . -name "*.rar" -exec unrar x -o+ {} \; +python -m venv venv && source venv/bin/activate +pip install -e . ``` -## Processing +## Memory -Processing the Collection 1 is much faster than the Collections 2-5. The estimates for Collections 2-5 are reported below. +Every command streams input and writes incrementally. Peak memory does not grow +with file size: a single 100 GB+ file is parsed with a small constant footprint. +`sort` and `clean` shell out to the OS `sort -u`, which spills to disk. -The parsing took around 20 hours on my server (CPU i7-8700K, 32GB of memory). I didn't have a large enough SSD to store all the temporary computations so everything -was done on a standard HDD. A faster disk will surely make the processing faster. +Measured on real Collections 2-5 data (19.94 GB, three files): **160 MB peak RSS** +for the whole process tree, versus **15.4 GB** for the original parser. Min RAM +available stayed at 15.3 GB, swap unchanged. -The sorting/removing duplicates step took 15 hours in total. +## Build the query dataset -The splitting into the smaller files (this file struct makes every query almost instantaneous) took a couple of hours at most. - -In total, expect around 2 days to process the Collections 2-5. +`build` is the resumable, disk-bounded pipeline. It parses the source in batches, +sorts+dedups each batch on a scratch disk, splits it into the bucket tree, then +dedups each bucket. Peak disk is source + tree, so it fits where a global sort +would not. ```bash -breach parse --path /path/to/extracted --success_file success.txt --failure_file failure.txt --cython_acceleration -rm -rf tmp && mkdir tmp # you need like 750GB in tmp/. By default /tmp/ is not enough for this! +collectionparser build \ + --source /path/to/extracted \ + --dataset /path/to/dataset \ + --scratch /path/to/scratch \ # fast disk for batch files + sort temp + --batch_gb 120 +``` + +**Pause / resume:** stop it any time (Ctrl-C or `systemctl stop`) and rerun the +same command. Finished batches are recorded in `build_state.json`, and parsed +files within the current batch in `progress_.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`. + +**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). + +## Processing (manual) + +```bash +collectionparser parse --path /path/to/extracted --success_file success.txt --failure_file failure.txt + +rm -rf tmp && mkdir tmp # sort needs room for temp files; /tmp is usually too small cat success.txt | pv -cN cut | sort -T tmp -u -S 90% --parallel=12 | pv -cN cut > success_sorted.txt -breach split --file success_sorted.txt --out data + +collectionparser split --file success_sorted.txt --out data ``` -## Converting the format of BreachCompilation to the new format +## Merging datasets -The dataset is available here: https://github.com/philipperemy/tensorflow-1.4-billion-password-analysis - -It's easy to convert the large `BreachCompilation` dataset to this format by running those commands. - -Expect those commands to take some time (less than a day). +`merge` is destructive: back up the destination first. ```bash -find /path/to/BreachCompilation/ -type f -exec cat {} + > breach_compilation.txt -rm -rf tmp && mkdir tmp # By default /tmp/ is not enough for this! -cat breach_compilation.txt | pv -cN cut | sort -T tmp -u -S 90% --parallel=12 | pv -cN cut > breach_compilation_sorted.txt -breach split --file breach_compilation_sorted.txt --out data_breach_compilation_sorted -``` - -From there a simple `breach merge` will be enough to merge it to the Collections 1 & 2 to 5. - -## Merging all the datasets together - -Run the Collection 1 and Collections 2-5 through the processing step described above. - -You will have two directories: `/path/to/collections1_data` and `/path/to/collections2_5_data`. - -Additionally, if you have the other dataset BreachCompilation, you will have another directory `/path/to/data_breach_compilation_sorted`, generated by the step above. - -The merge is destructive so it's better to create a copy of the output first and then merge each one into the output. - -``` cp -rf /path/to/collections1_data /path/to/big_dataset -breach merge --src /path/to/collections2_5_data --dest /path/to/big_dataset -breach merge --src /path/to/data_breach_compilation_sorted --dest /path/to/big_dataset +collectionparser merge --src /path/to/collections2_5_data --dest /path/to/big_dataset ``` -## Usage +## Query and export -The manual of the command line tool can be fetched by running `breach dumphelp`. +```bash +# exact (O(1) bucket lookup) or wildcard (* and ?): +collectionparser query --dataset data --pattern 'john*' +collectionparser query --dataset data --pattern '*@gmail.com' +# JSON Lines for OpenSearch / bulk ingest: +collectionparser export --dataset data --out records.jsonl ``` -Usage: cli [OPTIONS] COMMAND [ARGS]... -Options: - --debug / --no-debug - --help Show this message and exit. +## Measured performance -Commands: - chunk chunk large TXT files into smaller files. - clean Cleans a query friendly folder PATH. Move incorrect records and - sort the files. - dumphelp - evaluate Evaluates some metrics such as precision/recall (e.g. is OLD into - NEW). - merge Merges dataset SRC into dataset DEST. - parse 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. - sort Sorts a query friendly folder PATH. Target is itself. - split Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). - Use RESTART_FROM to resume from the i-th line. - test Infers passwords of a list of emails defined in FILE with a query - friendly folder DATASET. +Single large-file test on Collections 2-5 (3 files, 19.94 GB, 3 workers): -Usage: cli dumphelp [OPTIONS] +| metric | value | +| --- | --- | +| parse throughput | 20.5 MB/s aggregate (6.8 MB/s per worker) | +| valid pairs | 628,342,073 | +| rejects | 3,321,666 | +| peak RSS | 160 MB | +| `sort -u` | ~125 MB/s | +| `split` (sorted input) | ~43 MB/s | -Options: - --help Show this message and exit. +Split is ~12× slower on unsorted input (open-handle thrash); always sort first. -Usage: cli split [OPTIONS] +## Commands -Options: - --file FILE [required] - --out DIRECTORY [required] - --restart_from INTEGER [default: 0] - --help Show this message and exit. +Run `collectionparser --help` or `collectionparser --help`. -Usage: cli chunk [OPTIONS] - -Options: - --path DIRECTORY [required] - --size INTEGER [default: 50] - --help Show this message and exit. - -Usage: cli sort [OPTIONS] - -Options: - --path DIRECTORY [required] - --help Show this message and exit. - -Usage: cli clean [OPTIONS] - -Options: - --path DIRECTORY [required] - --help Show this message and exit. - -Usage: cli test [OPTIONS] - -Options: - --file FILE [required] - --dataset [breach_compilation|collections_1|collections_2_5|all] - [required] - --help Show this message and exit. - -Usage: cli parse [OPTIONS] - -Options: - --path DIRECTORY [required] - --success_file FILE [required] - --failure_file FILE [required] - --cython_acceleration / --no-cython_acceleration - --help Show this message and exit. - -Usage: cli merge [OPTIONS] - -Options: - --src DIRECTORY [required] - --dest DIRECTORY [required] - --help Show this message and exit. - -Usage: cli evaluate [OPTIONS] - -Options: - --old DIRECTORY [required] - --new DIRECTORY [required] - --help Show this message and exit. - ``` +| command | description | +| --- | --- | +| `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. | +| `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. | +| `clean` | Move misplaced records to the right bucket and sort. | +| `merge` | Merge one query-friendly folder into another. | +| `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. | diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..d3689db --- /dev/null +++ b/SKILL.md @@ -0,0 +1,111 @@ +--- +name: collectionparser +description: Parse, index, merge, and query multi-billion-row email:password breach collections (Collections 1-5, AntiPublic, Breach Compilation). Use when a task involves huge credential dumps, building a query-friendly a/b/c index, cracking a list of emails, or merging/sorting such datasets. Memory-safe on 100GB+ files. +--- + +# collectionparser + +Turns a folder of raw credential dumps into an `a/b/c` file tree that can be +queried in O(1), then looks up passwords for a list of emails. + +Install (Python 3.11+): + +```bash +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`). + +## How to run it for the user + +1. Find the input: either a single raw/sorted text file, or a folder of files. + Confirm the path — never guess and never write into the user's dataset + before backing up. +2. Run the pipeline below, one step at a time, watching for the sanity checks. +3. Report the output paths and the row counts printed by each step. + +## Pipeline (the intended order) + +```bash +# 1. Raw dumps in a folder -> valid (email:password) and rejected lines. +collectionparser parse --path /path/to/raw --success_file success.txt --failure_file failure.txt + +# 2. Sort + dedupe the whole success file. Keep tmp on a big disk. +rm -rf tmp && mkdir tmp +sort -T tmp -u -S 90% --parallel=12 success.txt > success_sorted.txt + +# 3. Sorted unique file -> query-friendly a/b/c tree. +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. +collectionparser test --file emails.txt --dataset data +``` + +To add another dataset to an existing index: + +```bash +cp -rf data big_dataset # merge is destructive; work on a copy +collectionparser merge --src other_data --dest big_dataset +``` + +## Commands + +| Command | Purpose | Key options | +| --- | --- | --- | +| `build` | Batched, resumable parse + dedup + split into the query tree. | `--source`, `--dataset`, `--scratch`, `--batch_gb` | +| `query` | Exact or wildcard (`*`, `?`) search over a dataset. | `--dataset`, `--pattern` | +| `export` | Export a dataset as JSON Lines (OpenSearch). | `--dataset`, `--out` | +| `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) | +| `sort` | `sort -u` every file in place. | `--path` | +| `clean` | Move misplaced records, then sort each file. | `--path` | +| `merge` | Merge `--src` into `--dest` (destructive). Runs `clean` after. | `--src`, `--dest` | +| `test` | Print passwords found for the emails in a file. | `--file`, `--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). + +## 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_.txt`. Stats land in `--scratch/stats.md`. +- **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. +- **`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`. + +## Verify the environment + +```bash +collectionparser --help +# round trip on toy data (~1s) +printf 'a@x.com:pw\nb@y.com:pw\n' > /tmp/raw.txt +collectionparser filter-lines --file /tmp/raw.txt --out /tmp/o.txt --less_than 1 --more_than 50 +``` + +If that works, the package and its entry point are installed correctly. diff --git a/breach/api.py b/breach/api.py deleted file mode 100644 index 0a7a605..0000000 --- a/breach/api.py +++ /dev/null @@ -1,154 +0,0 @@ -import os -import random -import shutil -import string -import subprocess -from glob import glob -from pathlib import Path - -from tqdm import tqdm - -from breach.parser import parse_to_files -from breach.query import QueryAPI, get_query_manager -from breach.splitter import get_output_file, write_line_to_file -from breach.utils import alpha_num_lookup, shuffle, read_lines, create_dir_for_file - - -class API: - - @staticmethod - def evaluate(old, new): - hit_count = 0 - total_count = 0 - files = [p for p in Path(old).expanduser().glob('**/*') if p.is_file()] - for i in range(1000): - lines = read_lines(random.choice(files)) - shuffle(lines) - random_lines = lines[0:10] - for random_line in random_lines: - password_list = QueryAPI.query(new, random_line.split(':')[0], 'email') - total_count += 1 - if len(password_list) == 0: - print(f'OLD: {random_line}, NEW: miss, HIT_RATE = {hit_count / total_count:.3f}.') - else: - hit_count += 1 - print(f'OLD: {random_line}, NEW: hit, HIT_RATE = {hit_count / total_count:.3f}.') - - @staticmethod - def merge(src, dest): - src_files = find_files(src) - dest_files = find_files(dest) - for src_index, src_file in src_files.items(): - if src_index in dest_files: - merge(src_file, dest_files[src_index]) - else: - move(src_file, dest + src_index) - - @staticmethod - def parse(path, success_file, failure_file, cython_acceleration=False): - parse_to_files(path, success_file, failure_file, cython_acceleration) - - @staticmethod - def sort(path): - path = str(Path(path).expanduser()) - paths = [Path(a) for a in glob(path + '/**/*', recursive=True) if Path(a).is_file() - and Path(a).suffix != '.sorted'] - for path in tqdm(paths, desc='Unique > Sorting > Replacing existing files'): - output_path = path.with_suffix('.sorted') - lines = read_lines(path) - unique_sorted_lines = sorted(set(lines)) - with open(output_path, 'w', encoding='utf8') as w: - w.write('\n'.join(unique_sorted_lines) + '\n') - shutil.move(output_path, path) - - @staticmethod - def clean(path): - files = list(find_files(Path(path).expanduser()).values()) - shuffle(files) - alpha = alpha_num_lookup() - num_files_moved = 0 - files_handlers = {} # u limit here! - with tqdm(files, desc='clean') as bar: - for file in bar: - lines_to_delete = [] - lines = read_lines(file) - for line in lines: - line_split = line.split(':') - ground_truth_file = get_output_file(alpha, line_split[0]) - key = str(file).replace(path, '') - if key.startswith('/'): - key = key[1:] - if ground_truth_file != key: - output_file = Path(path) / ground_truth_file - # print(f'MOVE: {line_split[0]} -> {output_file}.') - write_line_to_file(files_handlers, line_split, output_file, buffer_scale=10) - lines_to_delete.append(line) - num_files_moved += 1 - if num_files_moved % 100_000 == 0: - truncated_file = str(file).replace(path, "") - truncated_output_file = str(output_file).replace(path, "") - bar.set_description(f'clean. {num_files_moved:,} moved. ' - f'move: {line_split[0]}: {truncated_file}' - f' -> {truncated_output_file}') - lines2 = sorted(set(lines) - set(lines_to_delete)) - if len(lines) != len(lines2): - with open(str(file), 'w', encoding='utf8') as w: - w.write('\n'.join(lines2) + '\n') - - @staticmethod - def test(file: str, dataset: str): - qm = get_query_manager(dataset) - cracked_rate = 0 - lines = read_lines(file) - for line in lines: - results = qm.perform_query(line.strip(), 'email') - if len(results) > 0: - passwords = [p.split(':', 1)[-1] for p in results] - print('FOUND:', line, ':'.join(passwords)) - cracked_rate += 1 - print(f'Cracked rate: {cracked_rate / len(lines)}') - - @staticmethod - def chunk_files(path: str, max_size: int): - # split -b 50000000 file.txt file.txt. - # -d to have .00 and .01 is not compatible with MacOS... - files = list(find_files(Path(path).expanduser()).values()) - 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)) - subprocess.check_output(cmd) - f.unlink() - - @staticmethod - def filter_lines(file, out, less_than, more_than): - create_dir_for_file(out) - with open(out, 'w', encoding='utf8') as w: - with open(file, 'r', encoding='utf8', errors='ignore') as r: - for line in r: - line2 = ''.join(list(filter(lambda x: x in string.printable, line))) - if less_than <= len(line2.strip()) <= more_than: - w.write(line2) - - -def find_files(path: str): - path2 = Path(path).expanduser() - files = [p for p in path2.glob('**/*') if p.is_file()] - index = {str(f).replace(str(path2), ''): f for f in files} - assert len(index) == len(files) - return index - - -def merge(src, dst): - print(f'merge {src} {dst}') - with open(src, 'r', encoding='utf8', errors='ignore') as r: - content1 = r.read() - with open(dst, 'a', encoding='utf8', errors='ignore') as w: - w.write(content1) - print(f'del {src}') - os.remove(src) - - -def move(src, dst): - print(f'mv {src} {dst}') - shutil.move(src, dst) diff --git a/breach/cli.py b/breach/cli.py deleted file mode 100644 index a927fe7..0000000 --- a/breach/cli.py +++ /dev/null @@ -1,111 +0,0 @@ -import logging -from pathlib import Path - -import click - -from breach.api import API -from breach.splitter import split_file -from breach.utils import Ct, num_open_files_limit - - -# Extraction: nohup find . -name '*.tar.gz' -execdir tar --keep-newer-files -xzvf '{}' -C ../extracted \; & - -def recursive_help(cmd, parent=None): - ctx = click.core.Context(cmd, info_name=cmd.name, parent=parent) - print(cmd.get_help(ctx)) - print() - commands = getattr(cmd, 'commands', {}) - for sub in commands.values(): - recursive_help(sub, ctx) - - -@click.group() -@click.option('--debug/--no-debug') -@click.pass_context -def cli(ctx, debug): - if debug: - click.echo('Debugging mode enabled.') - logging.basicConfig(format='%(asctime)12s - %(levelname)s - %(message)s', level=logging.INFO) - 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') - exit(1) - - -@cli.command() -def dumphelp(): - recursive_help(cli) - - -@cli.command(short_help='Converts a large FILE to a query friendly folder OUT (e.g. a/b/c). ' - 'Use RESTART_FROM to resume from the i-th line.') -@click.option('--file', required=True, type=Ct.input_file()) -@click.option('--out', required=True, type=Ct.output_dir()) -@click.option('--restart_from', default=0, type=int, show_default=True) -def split(file, out, restart_from): - split_file(file, out, restart_from) - - -@cli.command(short_help='chunk large TXT files into smaller files.') -@click.option('--path', required=True, type=Ct.input_dir()) -@click.option('--size', default=50, type=int, show_default=True) # MB. -def chunk(path, size): - API.chunk_files(path, size) - - -@cli.command(short_help='Sorts a query friendly folder PATH. Target is itself.') -@click.option('--path', required=True, type=Ct.input_dir()) -def sort(path): - API.sort(path) - - -@cli.command(short_help='Cleans a query friendly folder PATH. Move incorrect records and sort the files.') -@click.option('--path', required=True, type=Ct.input_dir()) -def clean(path): - API.clean(path) - - -@cli.command(short_help='Infers passwords of a list of emails defined in FILE with a query friendly folder DATASET.') -@click.option('--file', required=True, type=Ct.input_file()) -# @click.option('--dataset', required=True, type=click.Choice(DATASET_CHOICES, case_sensitive=False)) -@click.option('--dataset', required=True) -def test(file: str, dataset: str): - API.test(file, dataset) - - -@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=Ct.output_dir()) -@click.option('--success_file', required=True, type=Ct.output_file()) -@click.option('--failure_file', required=True, type=Ct.output_file()) -@click.option('--cython_acceleration/--no-cython_acceleration', default=False) -def parse(path, success_file, failure_file, cython_acceleration): - API.parse(Path(path).expanduser(), success_file, failure_file, cython_acceleration) - - -@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.') -@click.option('--src', required=True, type=Ct.input_dir()) -@click.option('--dest', required=True, type=Ct.input_dir()) -def merge(src: str, dest: str): - API.merge(src, dest) - API.clean(dest) - - -@cli.command(short_help='Evaluates some metrics such as precision/recall (e.g. is OLD into NEW).') -@click.option('--old', required=True, type=Ct.input_dir()) -@click.option('--new', required=True, type=Ct.input_dir()) -def evaluate(old: str, new: str): - API.evaluate(old, new) - - -@cli.command(short_help='Filters lines less than N and more than M.') -@click.option('--file', required=True, type=Ct.input_file()) -@click.option('--out', required=True, type=Ct.output_file()) -@click.option('--less_than', required=True, type=int) -@click.option('--more_than', required=True, type=int) -def filter_lines(file: str, out: str, less_than: int, more_than: int): - API.filter_lines(file, out, less_than, more_than) - - -if __name__ == '__main__': - cli() diff --git a/breach/cparser.pyx b/breach/cparser.pyx deleted file mode 100644 index 991c03f..0000000 --- a/breach/cparser.pyx +++ /dev/null @@ -1,179 +0,0 @@ -import re -import string - -import cython -from validate_email import validate_email - -from breach.utils import PasswordType - -cdef inline c_is_md5(str password): - if len(password) == 32: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - -cdef inline c_is_sha1(str password): - if len(password) == 40: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - -cdef inline c_is_sha_256(str password): - if len(password) == 64: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - -cdef inline c_is_b_crypt(str password): - if 50 < len(password) < 70: - b_crypt_pattern_cond = password[0:4] == "$2a$" or password[0:4] == "$2b$" or password[0:4] == "$2y$" - if b_crypt_pattern_cond: - return True - return False - -cdef inline c_get_password_type(str password): - if c_is_md5(password): - return PasswordType.MD5 - elif c_is_sha1(password): - return PasswordType.SHA1 - elif c_is_sha_256(password): - return PasswordType.SHA256 - elif c_is_b_crypt(password): - return PasswordType.B_CRYPT - else: - return PasswordType.DECRYPTED - -@cython.boundscheck(False) -def c_parseline(str s): - # case mail@mail.com:password - # case mail@mail.com;password - # case mail@mail.com password - valid_separators = [':', ';', ' ', '\t', ' ', ' ', ', ', ',', ';;', '::', ','] - # https://github.com/swiftmailer/swiftmailer/issues/239 - b = s.strip() - b = ''.join(list(filter(lambda x: x in string.printable, b))) - while '..' in b: # TODO: could be move just for the email. - b = b.replace('..', '') - if b.startswith('mailto:'): - b = b[len('mailto:'):] - if b.startswith('E-mail:'): - b = b[len('E-mail:'):] - if b.startswith(':'): - b = b[1:] - if b.endswith(':'): - b = b[:-1] - - # TODO: could be move just for the email. but we dont need to be 100% accurate here. - b = b.replace('.@', '@') # just typo. - b = b.replace('@.', '@') # just typo. - b = b.replace(' @ ', '@') # just typo. - b = b.replace('@@', '@') # just typo. - - if ':' in b and ';' in b: - b = b.replace(';', ':') - - # FLIP=FALSE - for valid_separator in valid_separators: - result = c_parse_line(b, valid_separator, flip=False) - if result is not None: - return result - # FLIP=TRUE - for valid_separator in valid_separators: - result = c_parse_line(b, valid_separator, flip=True) - if result is not None: - return result - # print(f'DISCARD: {s.strip()}') - return None - -@cython.boundscheck(False) -cdef inline c_our_validate_email(str candidate): - parts = candidate.split('@') - if len(parts) != 2: - return False - if '.' not in parts[-1]: # .com or .fr - return False - return validate_email(candidate) - -@cython.boundscheck(False) -cdef c_parse_line(str row, str separator, bint flip=False): - cdef bint is_sep_in_row = separator in row - cdef bint sep_count_cond = False - cdef str mail = '' - cdef str password = '' - if not is_sep_in_row: # return fast. - return None - if separator != ' ': - parts = row.split(separator) - sep_count_cond = len(parts) == 2 - if len(parts) > 2: - row = separator.join(parts[-2:]) - # sh_karimian@yahoo.com:kiyan1365:16 - # wgwendlee@aol.com:godis2good: - sep_count_cond = True - elif is_sep_in_row: - sep_count_cond = True - else: - sep_count_cond = False - if sep_count_cond: - mail, password = row.split(separator, 1) - if flip: - password, mail = mail, password - password = password.replace(' ', '') - if len(password) > 30: # Password of 30 chars look like this: adfsfdsgdfghkfdghdsdfgdfgdfgdf. - return None - if len(password) == 0: - return None - if '\n\n' in password: - return None - if set(password) == {'?'}: # - gauntletskd@gmail.com:?????? ??? ????? - return None - if password.count('.') >= 3 and \ - re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', password) is not None: - return None - mail = mail.replace(' ', '') - # https://grox.net/utils/encoding.html - if '@' not in mail: - mail = mail.replace('%40', '@') - # - thakhuy.step129@hotmail.com.kkk111:125.26.106.158,nopparat - # count first because it's faster than compiling a regexp. - if len(mail) == 0: - return None - last_char = mail[-1] - first_char = mail[0] - # misseme21@yahoo.jp\ ilovebois3 - if last_char in ['<', '.', '\\', '>', '_']: - mail = mail[:-1] - # .y.heine@gmail.com atohiyamamot [exists] - if first_char in ['.', '<']: - mail = mail[1:] - if mail.endswith('@com'): - # iemongozalu@gmail@com mizo1210 - mail = mail.replace('@com', '.com') - if not c_our_validate_email(mail): - # some people replace neanta,buru@ezweb.ne,jp - mail = mail.replace(',', '.') - parts = mail.split('@') - if len(parts) > 2: - mail = '@'.join(parts[0:2]) - # TODO: do it for most common domain names. - if mail.count('@hotmail.com') == 2: # taa06@hotmail.com@hotmail.com:8765432f - mail = mail.replace('@hotmail.com', '', 1) - if c_our_validate_email(mail) and not c_our_validate_email(password): - password_type = c_get_password_type(password) - return mail, password, password_type - return None diff --git a/breach/parser.py b/breach/parser.py deleted file mode 100644 index 6677b30..0000000 --- a/breach/parser.py +++ /dev/null @@ -1,314 +0,0 @@ -import os -import re -import string -from glob import glob -from pathlib import Path -from time import time - -from breach.cparser import c_parseline -from tqdm import tqdm -from validate_email import validate_email - -from breach.utils import chunks, read_lines, PasswordType -from breach.utils import parallel_function, shuffle - - -# slow functions. - -def our_validate_email(candidate: str): - if candidate.count('@') != 1: - return False - if '.' not in candidate.split('@')[-1]: # .com or .fr - return False - return validate_email(candidate) - - -class LineParser: - - @staticmethod - def _parse_line(row, separator, flip=False): - if separator not in row: # return fast. - return None - if separator != ' ': - parts = row.split(separator) - sep_count_cond = len(parts) == 2 - if len(parts) > 2: - row = separator.join(parts[-2:]) - # sh_karimian@yahoo.com:kiyan1365:16 - # wgwendlee@aol.com:godis2good: - sep_count_cond = True - elif separator in row: - sep_count_cond = True - else: - sep_count_cond = False - if sep_count_cond: - mail, password = row.split(separator, 1) - if flip: - password, mail = mail, password - password = password.replace(' ', '') - if len(password) > 30: # Password of 30 chars look like this: adfsfdsgdfghkfdghdsdfgdfgdfgdf. - return None - if len(password) == 0: - return None - if '\n\n' in password: - return None - if set(password) == {'?'}: # - gauntletskd@gmail.com:?????? ??? ????? - return None - if password.count('.') >= 3 and \ - re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', password) is not None: - return None - mail = mail.replace(' ', '') - # https://grox.net/utils/encoding.html - if '@' not in mail: - mail = mail.replace('%40', '@') - # - thakhuy.step129@hotmail.com.kkk111:125.26.106.158,nopparat - # count first because it's faster than compiling a regexp. - - if len(mail) == 0: - return None - - last_char = mail[-1] - first_char = mail[0] - # misseme21@yahoo.jp\ ilovebois3 - if last_char in ['<', '.', '\\', '>', '_']: - mail = mail[:-1] - # .y.heine@gmail.com atohiyamamot [exists] - if first_char in ['.', '<']: - mail = mail[1:] - if mail.endswith('@com'): - # iemongozalu@gmail@com mizo1210 - mail = mail.replace('@com', '.com') - if not our_validate_email(mail): - # some people replace neanta,buru@ezweb.ne,jp - mail = mail.replace(',', '.') - parts = mail.split('@') - if len(parts) > 2: - mail = '@'.join(parts[0:2]) - # TODO: do it for most common domain names. - if mail.count('@hotmail.com') == 2: # taa06@hotmail.com@hotmail.com:8765432f - mail = mail.replace('@hotmail.com', '', 1) - if our_validate_email(mail) and not our_validate_email(password): - password_type = LineParser.get_password_type(password) - return mail, password, password_type - return None - - @staticmethod - def is_md5(password): - if len(password) == 32: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - - @staticmethod - def is_sha1(password): - if len(password) == 40: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - - @staticmethod - def is_sha_256(password): - if len(password) == 64: - for e in password: - if not ((ord('0') <= ord(e) <= ord('9')) or - (ord('A') <= ord(e) <= ord('F')) or - (ord('a') <= ord(e) <= ord('f'))): - return False - return True - else: - return False - - @staticmethod - def is_b_crypt(password): - b_crypt_pattern_cond = password[0:4] == "$2a$" or password[0:4] == "$2b$" or password[0:4] == "$2y$" - if 50 < len(password) < 70 and b_crypt_pattern_cond: - return True - return False - - @staticmethod - def get_password_type(password): - if LineParser.is_md5(password): - return PasswordType.MD5 - elif LineParser.is_sha1(password): - return PasswordType.SHA1 - elif LineParser.is_sha_256(password): - return PasswordType.SHA256 - elif LineParser.is_b_crypt(password): - return PasswordType.B_CRYPT - else: - return PasswordType.DECRYPTED - - @staticmethod - def parseline(s: str): - # case mail@mail.com:password - # case mail@mail.com;password - # case mail@mail.com password - valid_separators = [':', ';', ' ', '\t', ' ', ' ', ', ', ',', ';;', '::', ','] - # https://github.com/swiftmailer/swiftmailer/issues/239 - b = s.strip() - b = ''.join(list(filter(lambda x: x in string.printable, b))) - while '..' in b: # TODO: could be move just for the email. - b = b.replace('..', '') - if b.startswith('mailto:'): - b = b[len('mailto:'):] - if b.startswith('E-mail:'): - b = b[len('E-mail:'):] - if b.startswith(':'): - b = b[1:] - if b.endswith(':'): - b = b[:-1] - - # TODO: could be move just for the email. but we dont need to be 100% accurate here. - b = b.replace('.@', '@') # just typo. - b = b.replace('@.', '@') # just typo. - b = b.replace(' @ ', '@') # just typo. - b = b.replace('@@', '@') # just typo. - - if ':' in b and ';' in b: - b = b.replace(';', ':') - - # FLIP=FALSE - for valid_separator in valid_separators: - result = LineParser._parse_line(b, valid_separator, flip=False) - if result is not None: - return result - # FLIP=TRUE - for valid_separator in valid_separators: - result = LineParser._parse_line(b, valid_separator, flip=True) - if result is not None: - return result - # print(f'DISCARD: {s.strip()}') - return None - - -class Processor: - - def __init__(self, w=None, f=None, cython_acceleration=False): # for big files -> writes to disk directly. - if cython_acceleration: - print('Cython acceleration enabled.') - self.bind = c_parseline - else: - print('Cython acceleration disabled.') - self.bind = LineParser.parseline - self.w = w # cannot be serialized for multi processing. - self.f = f # cannot be serialized for multi processing. - - def process(self, filename: Path, is_big_file: bool = False): - success_list = [] - failure_list = [] - with open(str(filename), 'r', encoding='utf8', errors='ignore') as r: - buf_size = 10 ** 6 # for very large files. - while True: - lines = r.readlines(buf_size) - if len(lines) == 0: - break - for line in lines: - result = self.bind(line) - if result is not None: - mail, password, password_type = result - if password_type == PasswordType.DECRYPTED: - # mail is case insensitive. - success_str = ':'.join([str(mail).lower().strip(), password.strip()]) + '\n' - if is_big_file: - self.w.write(success_str) - else: - success_list.append(success_str) - else: - # stop writing binary data to file... - fail_line = ''.join(list(filter(lambda x: x in string.printable, line))) - if is_big_file: - self.f.write(fail_line) - else: - failure_list.append(fail_line) - return success_list, failure_list - - -def parse_to_files(input_dir, success_filename, failure_filename, cython_acceleration): - files = [p for p in Path(input_dir).glob('**/*') if p.is_file()] - shuffle(files) - # restart from where we stopped. - progress_filename = 'progress.txt' - if os.path.exists(progress_filename): - lines = read_lines(progress_filename) - else: - lines = [] - lines = {n: 1 for n in lines} - files = [f for f in files if str(f) not in lines] - shuffle(files) - # files.sort(key=lambda f: os.stat(f).st_size) # smallest to largest - - success_total = 0 - failure_total = 0 - num_threads = os.cpu_count() - start_time = time() - with open(progress_filename, 'a+') as p: - with open(failure_filename, 'a+', encoding='utf8') as f: - with open(success_filename, 'a+', encoding='utf8') as w: - large_files_processor = Processor(w=w, f=f, cython_acceleration=cython_acceleration) - small_files_processor = Processor(cython_acceleration=cython_acceleration) - with tqdm(files) as bar: - for chunk in chunks(files, num_threads): - # more than 300MB? - any_large_files = any([c.stat().st_size / 1e6 > 300 for c in chunk]) - if any_large_files: - # single thread and write directly to the files. - for chunky in chunk: - large_files_processor.process(chunky) - else: - # as many threads as we want. - results = parallel_function(small_files_processor.process, chunk, len(chunk)) - - for single_result in results: - success_list, failure_list = single_result - - for single_success in success_list: - w.write(single_success) - for single_success in failure_list: - f.write(single_success) - - success_total += len(success_list) - failure_total += len(failure_list) - - bar.update(len(chunk)) - bar.set_description(f'success = {success_total:,} failure = {failure_total:,}') - for single_chunk in chunk: - p.write(str(single_chunk) + '\n') - print(f'Time elapsed {time() - start_time:.3f} seconds.') - - -def statistics(input_dir): - files = glob(os.path.expanduser(input_dir) + '/**/*.*', recursive=True) - files = [f for f in files if not f.endswith('.tar.gz')] - shuffle(files) - parsed_lines = 0 - with open('stats.txt', 'w') as w: - w.write(','.join(['success', 'failure', 'failure_rate', 'filename']) + '\n') - for filename in files: - success = 0 - failure = 0 - lines = read_lines(filename) - parsed_lines += len(lines) - for line in lines: - result = LineParser.parseline(line) - if result is not None: - success += 1 - else: - failure += 1 - try: - failure_rate = failure / (success + failure) * 100 - except ZeroDivisionError: - failure_rate = 0.0 - w.write(','.join([str(success), str(failure), f'{failure_rate:.3f}', filename]) + '\n') - print(f'Parsed lines = {parsed_lines:,}.') - print(f'filename = {filename}, success = {success:,}, ' - f'failure = {failure:,}, failure rate = {failure_rate:.3f}%') diff --git a/breach/query.py b/breach/query.py deleted file mode 100644 index a7bdb80..0000000 --- a/breach/query.py +++ /dev/null @@ -1,194 +0,0 @@ -import logging -import os -import subprocess -from pathlib import Path - -logger = logging.getLogger(__name__) - - -# https://pypi.org/project/ripgrepy/ - -class BreachCompilationConstants: - EMAIL_INDEX = '/media/philippe/DATA/datasets/BreachCompilation/data' - PASSWORD_INDEX = '/media/philippe/DATA/datasets/BreachCompilation/passwords' - - -class Collections1: - EMAIL_INDEX = '/media/philippe/8TB/datasets/passwords2/data' - PASSWORD_INDEX = None - - -class Collections_2_5: - EMAIL_INDEX = '/media/philippe/8TB/datasets/passwords2/data' - PASSWORD_INDEX = None - - -DATASET_CHOICES = ['breach_compilation', 'collections_1', 'collections_2_5', 'all'] - - -def get_query_manager(dataset_choice: str): - if dataset_choice.lower() == 'breach_compilation': - return QueryManager(BreachCompilationConstants) - elif dataset_choice.lower() == 'collections_1': - return QueryManager(Collections1) - elif dataset_choice.lower() == 'collections_2_5': - return QueryManager(Collections_2_5) - elif os.path.exists(dataset_choice): - class Custom: - EMAIL_INDEX = dataset_choice - PASSWORD_INDEX = None - - return QueryManager(Custom) - else: - return QueryMultiManager() # all - - -def mask_pass_justin_rule(password: str, num_asterisks=-1) -> str: - if len(password) >= 8: - return password[0:2] + (len(password) - 3) * '*' + password[-1] - return password[0] + '*' * (len(password) - 2) + password[-1] - - -def mask_pass(password: str, num_asterisks=2) -> str: - if len(password) <= num_asterisks: - return '*' * len(password) - if len(password) >= 4: - i = len(password) // 2 - even = num_asterisks % 2 == 0 - half = num_asterisks // 2 - if even: - left = half - right = half - else: - left = half - right = half + 1 - return password[0:i - left] + '*' * num_asterisks + password[i + right:] - else: - return password - - -def extract_emails_and_passwords(txt_lines): - emails_passwords = [] - for txt_line in txt_lines: - try: - if '@' in txt_line: # does it contain an email address? - if all([char in txt_line for char in [':', ';']]): # which separator is it? : or ;? - separator = ':' - elif ':' in txt_line: # '_---madc0w---_@live.com:iskandar89 - separator = ':' - elif ';' in txt_line: # '_---lelya---_@mail.ru;ol1391ga - separator = ';' - else: - continue - - strip_txt_line = txt_line.strip() - email, password = strip_txt_line.split(separator) - emails_passwords.append((email, password)) - except Exception: - pass - return emails_passwords - - -def postprocess(results, num_asterisks_to_mask_password=0): - response = [b for b in extract_emails_and_passwords(results)] - if num_asterisks_to_mask_password > 0: - response = [(b[0], mask_pass_justin_rule(b[1])) for b in response] - response = [':'.join(b) for b in response] - response = sorted(set(response)) - return response - - -class QueryAPI: - - @staticmethod - def scanning(directory: Path): - file_list, dir_list = {}, {} - for x in directory.iterdir(): - if x.is_file(): - file_list[x.name] = x - else: - dir_list[x.name] = x - return file_list, dir_list - - @staticmethod - def search_in_file(path: Path, email): - assert path.is_file(), path - try: - results = subprocess.check_output(['rg', '--no-line-number', '-ie', '^' + - email, str(path)]).decode('utf8').split('\n') - results = [a for a in results if a != ''] - except subprocess.CalledProcessError: - results = [] - return results - - @staticmethod - def recursive_search(path: Path, characters: list, index: int): - fl, dl = QueryAPI.scanning(path) - cur = characters[index].lower() - if cur in fl: - return QueryAPI.search_in_file(fl[cur], ''.join(characters)) - elif cur in dl: - return QueryAPI.recursive_search(dl[cur], characters, index + 1) - elif 'symbols' in fl: - return QueryAPI.search_in_file(fl['symbols'], ''.join(characters)) - else: - raise Exception('Hello.') - - @staticmethod - def query(data_dir: str, request: str, query_type: str, num_asterisks_to_mask_password: int = 0): - if len(request) <= 3: - return [] - if query_type == 'email' and '@' not in request: - return [] - response = QueryAPI.recursive_search(Path(data_dir), list(request), 0) - return postprocess(response, num_asterisks_to_mask_password) - - -class QueryManagerStub: - def __init__(self, constants): - self.default_email_index = constants.EMAIL_INDEX - self.default_password_index = constants.PASSWORD_INDEX - self.counter = 0 - - def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0): - self.counter += 1 - if self.counter % 2 == 0: - return [f'{request}:hello123', f'{request}:hello123456'] - else: - return [] - - -class QueryManager: - - def __init__(self, constants): - self.default_email_index = constants.EMAIL_INDEX - self.default_password_index = constants.PASSWORD_INDEX - - def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0): - if query_type == 'email': - index = self.default_email_index - elif query_type == 'password': - index = self.default_password_index - else: - raise Exception({'error': 'Index is either email or password.'}) - results = [] - if ' ' in request: # batch. - for single_req in request.split(' '): - results.extend(QueryAPI.query(index, single_req, query_type, num_asterisks_to_mask_password)) - else: - results.extend(QueryAPI.query(index, request, query_type, num_asterisks_to_mask_password)) - return results - - -class QueryMultiManager: - - def __init__(self): - self.qm1 = QueryManager(BreachCompilationConstants) - self.qm2 = QueryManager(Collections1) - self.qm3 = QueryManager(Collections_2_5) - - def perform_query(self, request: str, query_type: str, num_asterisks_to_mask_password=0): - results = self.qm1.perform_query(request, query_type, num_asterisks_to_mask_password) - results.extend(self.qm2.perform_query(request, query_type, num_asterisks_to_mask_password)) - results.extend(self.qm3.perform_query(request, query_type, num_asterisks_to_mask_password)) - return sorted(set(results)) diff --git a/breach/splitter.py b/breach/splitter.py deleted file mode 100644 index 5bca487..0000000 --- a/breach/splitter.py +++ /dev/null @@ -1,80 +0,0 @@ -import csv -import io -import sys -from time import time - -from breach.utils import alpha_num_lookup, create_dir, create_dir_for_file, RingDictionary - -csv.field_size_limit(sys.maxsize) - - -def split_file(breach_compilation_file, - output_dir='out', - restart_from=0): - # It seems like this function is not 100% correct as we need to clean the dataset after this. - # It's not a big deal but I don't understand why we have to do that. - # Could probably be to a limitation to the large number of files we open at the same time x increase of buffer. - create_dir(output_dir) - alpha = alpha_num_lookup() - files_handlers = RingDictionary() - t1 = time() - with open(breach_compilation_file, 'r', encoding='utf8', errors='ignore') as r: - for i, line in enumerate(r): - if i % 10_000_000 == 0: - elapsed_time = time() - t1 - t1 = time() - print(f'|- Processed {i:,} rows in {elapsed_time:.2f} seconds.') - if i < restart_from: - continue - try: # faster than if: https://stackoverflow.com/questions/1835756/using-try-vs-if-in-python - email, password = line.strip().split(':', 1) - except ValueError: - continue - - output_file = get_output_file(alpha, email) - output_file = output_dir + '/' + output_file - write_line_to_file(files_handlers, line, output_file) - - -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) - except KeyError: - create_dir_for_file(output_file) - files_handlers[output_file] = open(file=output_file, - mode='a+', - encoding='utf8', - buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE) - files_handlers[output_file].write(line) - - -def get_output_file(alpha_lookup, email): - try: # most likely case. - letter_0 = email[0].lower() - letter_1 = email[1].lower() - letter_2 = email[2].lower() - except IndexError: - email_len = len(email) - if email_len == 0: - letter_0 = '' - letter_1 = '' - letter_2 = '' - elif email_len == 1: - letter_0 = email[0].lower() - letter_1 = '' - letter_2 = '' - elif email_len == 2: - letter_0 = email[0].lower() - letter_1 = email[1].lower() - letter_2 = '' - else: - raise Exception('Error in program.') - if letter_0 not in alpha_lookup: - output_file = 'symbols' - elif letter_1 not in alpha_lookup: - output_file = letter_0 + '/symbols' - elif letter_2 not in alpha_lookup: - output_file = letter_0 + '/' + letter_1 + '/symbols' - else: - output_file = letter_0 + '/' + letter_1 + '/' + letter_2 - return output_file diff --git a/breach/utils.py b/breach/utils.py deleted file mode 100644 index 4fc6859..0000000 --- a/breach/utils.py +++ /dev/null @@ -1,142 +0,0 @@ -import logging -import os -import random -import subprocess -from datetime import datetime -from multiprocessing import Pool - -import click - - -def shuffle(lst): - random.seed(123) - random.shuffle(lst) - - -def num_open_files_limit(): - return int(subprocess.check_output('ulimit -n', shell=True).decode().strip()) - - -def read_lines(filename: str): - with open(filename, 'r', encoding='utf8', errors='ignore') as r: - return r.read().strip().split('\n') - - -def create_dir(output_dir: str): - if len(output_dir) > 0 and not os.path.exists(output_dir): - os.makedirs(output_dir) - - -def create_dir_for_file(filename: str): - create_dir(os.path.dirname(filename)) - - -def alpha_num_lookup(): - alpha = list('abcdefghijklmnopqrstuvwxyz0123456789') - alpha_lookup = {a: 1 for a in alpha} - assert len(alpha) == 36 - return alpha_lookup - - -def init_logger(name): - format_str = '%(asctime)12s - %(levelname)s - %(message)s' - formatter = logging.Formatter(format_str) - log_dir = 'logs' - create_dir(log_dir) - log_filename = os.path.join(log_dir, name + '_' + datetime.now().strftime('%Y-%m-%d_%H:%M:%S') + '.log') - logging.basicConfig(format=format_str, filename=log_filename, level=logging.INFO) - console = logging.StreamHandler() - console.setFormatter(formatter) - logging.getLogger('').addHandler(console) - - -def parallel_function(f, sequence, num_threads=None): - if num_threads == 1: - return [f(x) for x in sequence] - pool = Pool(processes=num_threads) - result = pool.map(f, sequence) - cleaned = [x for x in result if x is not None] - pool.close() - pool.join() - return cleaned - - -def chunks(lst, n): - """Yield successive n-sized chunks from l.""" - for i in range(0, len(lst), n): - yield lst[i:i + n] - - -class Ct: - - @staticmethod - def input_file(writable=False): - return click.Path(exists=True, file_okay=True, dir_okay=False, - writable=writable, readable=True, resolve_path=True) - - @staticmethod - def input_dir(writable=False): - return click.Path(exists=True, file_okay=False, dir_okay=True, - writable=writable, readable=True, resolve_path=True) - - @staticmethod - def output_file(): - return click.Path(exists=False, file_okay=True, dir_okay=False, - writable=True, readable=True, resolve_path=True) - - @staticmethod - def output_dir(): - return click.Path(exists=False, file_okay=False, dir_okay=True, - writable=True, readable=True, resolve_path=True) - - -class PasswordType: - MD5 = 'MD5' - SHA1 = 'SHA1' - SHA256 = 'SHA256' - B_CRYPT = 'B_CRYPT' - DECRYPTED = 'DECRYPTED' - - -class RingDictionary: - - def __init__(self, max_num_keys=100, debug=False): - self.d = {} # output_file -> file_handler. - self.k = [] - self.max_num_keys = max_num_keys - self.debug = debug - - def __getitem__(self, key): - return self.d[key] - - def __del__(self): - for k in self.k: - self.d[k].flush() - if self.debug: - print(f'[RingDictionary] Flush stream: {k}.') - - def __setitem__(self, key, item): - try: - self.d[key] - except KeyError: - if len(self.k) >= self.max_num_keys: - first_item = self.k.pop(0) - try: - self.d[first_item].flush() # for streams. - if self.debug: - print(f'[RingDictionary] Flush stream: {first_item}.') - except Exception: - pass - del self.d[first_item] - if self.debug: - print(f'[RingDictionary] Drop: {first_item}.') - if self.debug: - print(f'[RingDictionary] Add: {key}.') - self.k.append(key) - self.d[key] = item - - def keys(self): - return self.d.keys() - - def values(self): - return self.d.values() diff --git a/breach/__init__.py b/collectionparser/__init__.py similarity index 100% rename from breach/__init__.py rename to collectionparser/__init__.py diff --git a/collectionparser/api.py b/collectionparser/api.py new file mode 100644 index 0000000..84e04c0 --- /dev/null +++ b/collectionparser/api.py @@ -0,0 +1,177 @@ +import json +import os +import random +import shutil +import subprocess +from pathlib import Path + +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) + +# ponytail: buckets under this size sort byte-exactly in RAM; larger buckets +# shell out to `sort -u`, which spills to disk. +_IN_MEMORY_SORT_LIMIT = 8 * 1024 * 1024 + + +class API: + + @staticmethod + def evaluate(old, new): + 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): + if count < 10: + random_lines.append(line.rstrip('\n')) + elif (j := random.randint(0, count)) < 10: + random_lines[j] = line.rstrip('\n') + for random_line in random_lines: + password_list = QueryAPI.query(new, random_line.partition(':')[0], 'email') + total_count += 1 + hit = 'hit' if password_list else 'miss' + hit_count += bool(password_list) + print(f'OLD: {random_line}, NEW: {hit}, HIT_RATE = {hit_count / total_count:.3f}.') + + @staticmethod + def merge(src, dest): + 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]) + else: + dst = dest_root / src_index.lstrip('/') + create_dir_for_file(str(dst)) + move(str(src_file), str(dst)) + + @staticmethod + def parse(path, success_file, failure_file): + parse_to_files(path, success_file, failure_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) + + @staticmethod + def clean(path): + files = find_files(Path(path).expanduser()) + shuffle(files) + num_files_moved = 0 + files_handlers = {} # u limit here! + 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: + 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) + 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, "")}') + else: + w.write(line) + sort_unique(unique_file, file) + os.remove(unique_file) + + @staticmethod + def test(file: str, dataset: str): + cracked_rate = 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 + total += 1 + print(f'Cracked rate: {cracked_rate / total if total else 0.0}') + + @staticmethod + def chunk_files(path: str, max_size: int): + # -d (numeric suffixes) is not compatible with MacOS. + 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)) + subprocess.run(cmd, check=True) + f.unlink() + + @staticmethod + def export_jsonl(dataset: str, out: str): + """Export the dataset as JSON Lines (`{"email":..., "password":...}`).""" + create_dir_for_file(out) + total = 0 + with open(out, 'w', encoding='utf8') as w: + 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: + email, _, password = line.rstrip('\n').partition(':') + w.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): + 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) + + +def find_files(path): + root = Path(path).expanduser() + return [p for p in root.glob('**/*') if p.is_file()] + + +def sort_unique(src, dst): + src = Path(src) + if src.stat().st_size <= _IN_MEMORY_SORT_LIMIT: + lines = src.read_bytes().split(b'\n') + if lines and lines[-1] == b'': + lines.pop() + 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, + env={**os.environ, 'LC_ALL': 'C'}) + + +def merge(src, dst): + print(f'merge {src} {dst}') + with open(src, 'rb') as r, open(dst, 'ab') as w: + shutil.copyfileobj(r, w, length=1024 * 1024) + print(f'del {src}') + os.remove(src) + + +def move(src, dst): + print(f'mv {src} {dst}') + shutil.move(src, dst) diff --git a/collectionparser/build.py b/collectionparser/build.py new file mode 100644 index 0000000..10e8ce6 --- /dev/null +++ b/collectionparser/build.py @@ -0,0 +1,271 @@ +"""Resumable batched build of the deduped query dataset. + +Why batches: the raw collection is ~908 GB but the volume plus a sort temp +cannot hold a global sort. Each batch is parsed, sorted+deduped on the scratch +disk, and split into the same bucket tree; a final per-bucket sort dedups +across batches. Peak disk = source + tree. + +Resume: completed batches are listed in ``build_state.json``; within a batch, +each parsed input file is recorded in ``progress_.txt``. Stop with Ctrl-C / +``systemctl stop`` and rerun the same command to continue. The parse phase +resumes per file; sort+split of an interrupted batch is redone (duplicates are +removed by the final dedup), so an interrupted split does not lose data. + +Archives are extracted into the scratch disk for the duration of their batch and +removed once parsed; the source is never modified. +""" + +import bz2 +import gzip +import json +import lzma +import os +import shutil +import subprocess +import tarfile +import time +import zipfile +from pathlib import Path + +from collectionparser.api import API +from collectionparser.parser import parse_files +from collectionparser.splitter import split_file + +DEFAULT_BATCH_BYTES = 120 * 1024 ** 3 +ARCHIVE_SUFFIXES = ('.rar', '.zip', '.7z', '.tar', '.tar.gz', '.tgz', '.tar.bz2', + '.tbz2', '.tar.xz', '.txz', '.gz', '.bz2', '.xz') +DEVNULL = subprocess.DEVNULL + + +def _is_archive(path: Path) -> bool: + return path.name.lower().endswith(ARCHIVE_SUFFIXES) + + +def _which(*names): + for name in names: + found = shutil.which(name) + if found: + return found + return None + + +def _safe_target(dest: Path, name: str) -> Path: + target = (dest / name).resolve() + if not str(target).startswith(str(dest.resolve()) + os.sep): + raise ValueError(f'unsafe archive member: {name}') + return target + + +def _extract_compressed(path: Path, dest: Path): + """Try the available RAR/7z extractors in turn. + + No single tool wins: `unrar` handles real RARs, `7z` handles the many + mislabeled archives in the wild. Try both, first rc in (0, 1) wins. + ponytail: 7z may follow archive paths outside dest; trusted local dumps only. + """ + seven = _which('7z', '7za') + unrar = _which('unrar') + if path.name.lower().endswith('.rar'): + candidates = [unrar, seven] + else: + candidates = [seven, unrar] + last = None + for tool in (c for c in candidates if c): + shutil.rmtree(dest, ignore_errors=True) + dest.mkdir(parents=True, exist_ok=True) + if 'unrar' in tool: + 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 + 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): + with tarfile.open(path) as t: + try: + t.extractall(dest, filter='data') # py3.12+ + except TypeError: # py3.11 + t.extractall(dest) + + +def _extract_archive(path: Path, dest: Path): + name = path.name.lower() + dest.mkdir(parents=True, exist_ok=True) + if name.endswith('.zip'): + with zipfile.ZipFile(path) as z: + for member in z.namelist(): + _safe_target(dest, member) + z.extractall(dest) + elif name.endswith(('.rar', '.7z')): + _extract_compressed(path, dest) + elif name.endswith(('.tar', '.tar.gz', '.tgz', '.tar.bz2', '.tbz2', '.tar.xz', '.txz')): + _extract_tar(path, dest) + else: # single-file .gz/.bz2/.xz + opener = gzip.open if name.endswith('.gz') else bz2.open if name.endswith('.bz2') else lzma.open + with opener(path, 'rb') as r, open(dest / path.stem, 'wb') as w: + 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()] + 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 + for f in files: + sz = f.stat().st_size + if batch and size + sz > batch_bytes: + yield batch + batch, size = [], 0 + batch.append(f) + size += sz + if batch: + yield batch + + +def _count_lines(path): + total = 0 + with open(path, 'rb') as r: + for chunk in iter(lambda: r.read(1 << 20), b''): + total += chunk.count(b'\n') + return total + + +def _stats_markdown(state): + stats = state['stats'] + total_in = sum(s['input_bytes'] for s in stats) + total_pairs = sum(s['pairs'] for s in stats) + total_rej = sum(s['rejects'] for s in stats) + total_s = sum(s['total_s'] for s in stats) + lines = [ + '# collectionparser build stats', + '', + f'- input: {total_in / 1e9:.1f} GB across {sum(s["files"] for s in stats):,} files', + f'- valid pairs: {total_pairs:,}', + f'- rejects: {total_rej:,}', + f'- batches: {len(stats)}', + f'- build seconds: {total_s:,.0f}', + ] + if total_s: + lines.append(f'- aggregate throughput: {total_in / 1e6 / total_s:.1f} MB/s') + 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: + lines.append(f'- unique records: {state["unique_records"]:,}') + return '\n'.join(lines) + '\n' + + +def _extract_batch(archives, extract_dir, index): + if not archives: + return [] + t0 = time.time() + failed = 0 + errors = [] + 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}') + 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 + + +def build(source, dataset, scratch, batch_bytes=DEFAULT_BATCH_BYTES, dedup=True, extract=True): + dataset = Path(dataset).expanduser() + scratch = Path(scratch).expanduser() + dataset.mkdir(parents=True, exist_ok=True) + scratch.mkdir(parents=True, exist_ok=True) + for stale in scratch.glob('extract_*'): + shutil.rmtree(stale, ignore_errors=True) + + if extract and not _which('7z', '7za', 'unrar'): + print('WARN: no 7z/7za/unrar found; .rar/.7z archives will be skipped ' + '(.zip/.tar/.gz still work).', flush=True) + + state_path = scratch / 'build_state.json' + state = json.loads(state_path.read_text()) if state_path.exists() else {'done': [], 'stats': []} + done = set(state['done']) + files = _discover(source) + batches = list(_batches(files, batch_bytes)) + print(f'{len(files)} files, {len(batches)} batches, {len(done)} already done.', flush=True) + + failures = scratch / 'failures.txt' + for index, batch in enumerate(batches): + if index in done: + continue + success = scratch / f'success_{index:05d}.txt' + failure = scratch / f'failure_{index:05d}.txt' + extract_dir = scratch / f'extract_{index:05d}' + progress_path = scratch / f'progress_{index:05d}.txt' + input_bytes = sum(f.stat().st_size for f in batch) + + if not success.exists(): + 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: + 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)) + 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)) + 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)}) + split_file(str(success), str(dataset)) + success.unlink(missing_ok=True) + if failure.exists(): + with open(failures, 'ab') as w, open(failure, 'rb') as r: + shutil.copyfileobj(r, w, length=1024 * 1024) + failure.unlink(missing_ok=True) + progress_path.unlink(missing_ok=True) + else: + pairs = rejects = 0 + shutil.rmtree(extract_dir, ignore_errors=True) + + elapsed = time.time() - t0 + stat = {'batch': index, 'files': len(batch), 'input_bytes': input_bytes, + 'pairs': pairs, 'rejects': rejects, 'total_s': round(elapsed, 1), + 'mb_per_s': round(input_bytes / 1e6 / elapsed, 2) if elapsed else 0.0} + state['stats'].append(stat) + done.add(index) + state['done'] = sorted(done) + state_path.write_text(json.dumps(state, indent=2)) + (scratch / 'stats.md').write_text(_stats_markdown(state)) + print('BATCH ' + json.dumps(stat), flush=True) + + 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) + state['deduped'] = True + state_path.write_text(json.dumps(state, indent=2)) + + if state.get('unique_records') is None: + state['unique_records'] = sum(_count_lines(p) for p in dataset.glob('**/*') if p.is_file()) + state_path.write_text(json.dumps(state, indent=2)) + + (scratch / 'stats.md').write_text(_stats_markdown(state)) + print('BUILD COMPLETE', flush=True) diff --git a/collectionparser/cli.py b/collectionparser/cli.py new file mode 100644 index 0000000..cff4b62 --- /dev/null +++ b/collectionparser/cli.py @@ -0,0 +1,128 @@ +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 \; & + + +@click.group() +@click.option('--debug/--no-debug') +@click.pass_context +def cli(ctx, 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). ' + 'Use RESTART_FROM to resume from the i-th line.') +@click.option('--file', required=True, type=input_file()) +@click.option('--out', required=True, type=output_dir()) +@click.option('--restart_from', default=0, type=int, show_default=True) +def split(file, out, restart_from): + split_file(file, out, restart_from) + + +@cli.command(short_help='chunk large TXT files into smaller files.') +@click.option('--path', required=True, type=input_dir()) +@click.option('--size', default=50, type=int, show_default=True) # MB. +def chunk(path, size): + API.chunk_files(path, size) + + +@cli.command(short_help='Sorts a query friendly folder PATH. Target is itself.') +@click.option('--path', required=True, type=input_dir()) +def sort(path): + API.sort(path) + + +@cli.command(short_help='Cleans a query friendly folder PATH. Move incorrect records and sort the files.') +@click.option('--path', required=True, type=input_dir()) +def clean(path): + API.clean(path) + + +@cli.command(short_help='Infers passwords of a list of emails defined in FILE with a query friendly folder DATASET.') +@click.option('--file', required=True, type=input_file()) +@click.option('--dataset', required=True) +def test(file: str, dataset: str): + API.test(file, dataset) + + +@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('--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) + + +@cli.command(short_help='Merges dataset SRC into dataset DEST. Duplicates are removed in the second pass.') +@click.option('--src', required=True, type=input_dir()) +@click.option('--dest', required=True, type=input_dir()) +def merge(src: str, dest: str): + API.merge(src, dest) + API.clean(dest) + + +@cli.command(short_help='Evaluates some metrics such as precision/recall (e.g. is OLD into NEW).') +@click.option('--old', required=True, type=input_dir()) +@click.option('--new', required=True, type=input_dir()) +def evaluate(old: str, new: str): + API.evaluate(old, new) + + +@cli.command(short_help='Filters lines less than N and more than M.') +@click.option('--file', required=True, type=input_file()) +@click.option('--out', required=True, type=output_file()) +@click.option('--less_than', required=True, type=int) +@click.option('--more_than', required=True, type=int) +def filter_lines(file: str, out: str, less_than: int, more_than: int): + API.filter_lines(file, out, less_than, more_than) + + +@cli.command(short_help='Searches a dataset by exact email or wildcard pattern ' + '(e.g. "*@gmail.com", "john*", "jo?n@*").') +@click.option('--dataset', required=True, type=input_dir()) +@click.option('--pattern', required=True) +def query(dataset: str, pattern: str): + if '*' in pattern or '?' in pattern: + results = QueryAPI.wildcard(dataset, pattern) + else: + results = QueryAPI.query(dataset, pattern, 'email') + if results: + click.echo('\n'.join(results)) + + +@cli.command(short_help='Exports a dataset as JSON Lines for OpenSearch/bulk ingest.') +@click.option('--dataset', required=True, type=input_dir()) +@click.option('--out', required=True, type=output_file()) +def export(dataset: str, out: str): + API.export_jsonl(dataset, out) + + +@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()) +@click.option('--dataset', required=True, type=output_dir()) +@click.option('--scratch', required=True, type=output_dir()) +@click.option('--batch_gb', default=120, type=int, show_default=True) +@click.option('--dedup/--no-dedup', default=True) +@click.option('--extract/--no-extract', default=True, show_default=True, + help='Extract archives (.rar/.zip/.7z/.tar.*/.gz) found in the source.') +def build(source: str, dataset: str, scratch: str, batch_gb: int, dedup: bool, extract: bool): + from collectionparser.build import build as run_build + run_build(source, dataset, scratch, batch_bytes=batch_gb * 1024 ** 3, + dedup=dedup, extract=extract) + + +if __name__ == '__main__': + cli() diff --git a/collectionparser/parser.py b/collectionparser/parser.py new file mode 100644 index 0000000..70ed96c --- /dev/null +++ b/collectionparser/parser.py @@ -0,0 +1,205 @@ +import os +import re +import shutil +import string +import uuid +from concurrent.futures import ProcessPoolExecutor +from glob import glob +from pathlib import Path +from time import time + +from tqdm import tqdm + +from collectionparser.utils import PasswordType, filter_printable, shuffle + +# Separators tried in order; the duplicate ',' of the original is dropped. +VALID_SEPARATORS = (':', ';', ' ', '\t', ' ', ' ', ', ', ',', ';;', '::') + +# Same shape as the old `validate_email` dependency: rejects '<', '/', '>' etc. +_EMAIL_RE = re.compile( + r"^[a-z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[a-z0-9!#$%&'*+/=?^_`{|}~-]+)*" + r"@(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$", + re.IGNORECASE) +_IP_RE = re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}') +_HEX = frozenset(string.hexdigits) +_HASH_TYPES = {32: PasswordType.MD5, 40: PasswordType.SHA1, 64: PasswordType.SHA256} + + +def valid_email(candidate: str) -> bool: + return candidate.count('@') == 1 and _EMAIL_RE.match(candidate) is not None + + +def password_type(password: str) -> PasswordType: + hash_type = _HASH_TYPES.get(len(password)) + if hash_type is not None and _HEX.issuperset(password): + return hash_type + if 50 < len(password) < 70 and password[:4] in ('$2a$', '$2b$', '$2y$'): + return PasswordType.B_CRYPT + return PasswordType.DECRYPTED + + +def parse_line(row, separator, flip=False): + if separator not in row: + return None + if separator != ' ': + parts = row.split(separator) + if len(parts) > 2: + # sh_karimian@yahoo.com:kiyan1365:16 -> keep the last two fields. + row = separator.join(parts[-2:]) + mail, password = row.split(separator, 1) + if flip: + mail, password = password, mail + password = password.replace(' ', '') + if not password or len(password) > 30 or '\n\n' in password: + return None + if set(password) == {'?'}: # gauntletskd@gmail.com:?????? ??? ????? + return None + if password.count('.') >= 3 and _IP_RE.search(password): + return None + mail = mail.replace(' ', '') + if '@' not in mail: # https://grox.net/utils/encoding.html + mail = mail.replace('%40', '@') + if not mail: + return None + last_char, first_char = mail[-1], mail[0] + # misseme21@yahoo.jp\ ilovebois3 / .y.heine@gmail.com atohiyamamot + if last_char in '<.\\>_': + mail = mail[:-1] + if first_char in '.<': + mail = mail[1:] + if mail.endswith('@com'): # iemongozalu@gmail@com mizo1210 + mail = mail.replace('@com', '.com') + if not valid_email(mail): + # some people replace neanta,buru@ezweb.ne,jp + mail = mail.replace(',', '.') + parts = mail.split('@') + if len(parts) > 2: + mail = '@'.join(parts[:2]) + # TODO: do it for most common domain names. + if mail.count('@hotmail.com') == 2: # taa06@hotmail.com@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 None + + +def parseline(s: str): + b = filter_printable(s.strip()) + while '..' in b: + b = b.replace('..', '') + b = b.removeprefix('mailto:').removeprefix('E-mail:') + b = b.removeprefix(':').removesuffix(':') + # TODO: could be moved just for the email, but we don't need 100% accuracy. + b = b.replace('.@', '@').replace('@.', '@').replace(' @ ', '@').replace('@@', '@') + if ':' in b and ';' in b: + b = b.replace(';', ':') + # A valid email needs an '@'; '%40' is the only cleanup that can create one. + # Skip the whole separator scan for lines that can never yield a pair. + if '@' not in b and '%40' not in b: + return None + for flip in (False, True): + for separator in VALID_SEPARATORS: + result = parse_line(b, separator, flip) + if result is not None: + return result + return None + + +class Processor: + """Streams one input file line by line, appending results straight to disk. + + Never holds parsed output in memory: workers append to the shared + success/failure files and return only (success, failure) counts, so memory + per worker is a single read buffer regardless of file size. + """ + + def __init__(self, success_filename=None, failure_filename=None): + self.success_filename = success_filename + self.failure_filename = failure_filename + + # Each worker writes its own buffered part file; the parent concatenates + # parts after every chunk. Buffered writes avoid the one-syscall-per-line + # cost of shared O_APPEND, and separate files keep concurrent lines from + # 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): + 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, \ + open(f'{self.success_filename}.part.{part}', 'w', encoding='utf8') as sw, \ + open(f'{self.failure_filename}.part.{part}', 'w', encoding='utf8') as fw: + for line in r: + result = parseline(line) + if result is None: + fw.write(filter_printable(line)) # stop writing binary data to file. + failure += 1 + elif result[2] == PasswordType.DECRYPTED: + mail, password, _ = result + sw.write(f'{str(mail).lower().strip()}:{password.strip()}\n') + success += 1 + return success, failure + + +def _clear_parts(*finals): + # 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: + for part in glob(f'{final}.part.*'): + os.remove(part) + + +def _merge_parts(final): + parts = glob(f'{final}.part.*') + if not parts: + return + with open(final, 'ab') as w: + for part in parts: + with open(part, 'rb') as r: + shutil.copyfileobj(r, w, length=1024 * 1024) + 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) + success_total = failure_total = 0 + workers = os.cpu_count() or 1 + processor = Processor(success_filename, failure_filename) + start_time = time() + progress = open(progress_filename, 'a+', encoding='utf8') if progress_filename else None + try: + with ProcessPoolExecutor(max_workers=workers) as executor, tqdm(files) as bar: + for i in range(0, len(files), workers): + chunk = files[i:i + workers] + for success, failure in executor.map(processor.process, chunk): + success_total += success + failure_total += failure + # Merge only after the whole chunk finished, then mark it done, so + # a crash never records a file whose output was not written. + _merge_parts(success_filename) + _merge_parts(failure_filename) + bar.update(len(chunk)) + bar.set_description(f'success = {success_total:,} failure = {failure_total:,}') + 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() + 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} + 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') diff --git a/collectionparser/query.py b/collectionparser/query.py new file mode 100644 index 0000000..2b17f2c --- /dev/null +++ b/collectionparser/query.py @@ -0,0 +1,105 @@ +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: + if '@' not in txt_line: # does it contain an email address? + continue + if ':' in txt_line: # '_---madc0w---_@live.com:iskandar89 + separator = ':' + elif ';' in txt_line: # '_---lelya---_@mail.ru;ol1391ga + separator = ';' + else: + continue + try: + 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)}) + + +def glob_to_regex(pattern: str) -> str: + """Glob -> regex anchored to the email field (`*`/`?` never cross the ':'). + + ponytail: `*` matches within the email only; a `*` that must cross the + separator needs a different index. + """ + parts = [] + for ch in pattern: + if ch == '*': + parts.append('[^:]*') + elif ch == '?': + parts.append('[^:]') + else: + parts.append(re.escape(ch)) + return '^' + ''.join(parts) + ':' + + +class QueryAPI: + + @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 + else: + dir_list[x.name] = x + return file_list, dir_list + + @staticmethod + def search_in_file(path: Path, email): + assert path.is_file(), path + try: + output = subprocess.check_output( + ['rg', '--no-line-number', '-ie', '^' + re.escape(email), str(path)], text=True) + except subprocess.CalledProcessError: + return [] + return [line for line in output.splitlines() if line] + + @staticmethod + def recursive_search(path: Path, characters: list, index: int): + 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) + 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): + if len(request) <= 3: + return [] + if query_type == 'email' and '@' not in request: + return [] + 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. + + ponytail: full scan of the tree (ripgrep is fast enough); add a + domain-sharded index only if wildcard latency becomes a problem. + """ + try: + output = subprocess.check_output( + ['rg', '--no-filename', '--no-line-number', '-i', '-e', + glob_to_regex(pattern), str(Path(data_dir))], text=True) + except subprocess.CalledProcessError: + return [] + return postprocess(output.splitlines()) diff --git a/collectionparser/splitter.py b/collectionparser/splitter.py new file mode 100644 index 0000000..313fbc2 --- /dev/null +++ b/collectionparser/splitter.py @@ -0,0 +1,47 @@ +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 + +_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. + 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): + if i % 10_000_000 == 0: + elapsed = time() - t1 + t1 = 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)}') + + +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) + except KeyError: + create_dir_for_file(output_file) + files_handlers[output_file] = open(output_file, mode='a+', encoding='utf8', + buffering=buffer_scale * io.DEFAULT_BUFFER_SIZE) + files_handlers[output_file].write(line) + + +@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) diff --git a/collectionparser/utils.py b/collectionparser/utils.py new file mode 100644 index 0000000..dd1966a --- /dev/null +++ b/collectionparser/utils.py @@ -0,0 +1,89 @@ +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 filter_printable(text: str) -> str: + return _NON_PRINTABLE.sub('', text) + + +def iter_lines(filename): + with open(filename, 'r', encoding='utf8', errors='ignore') as r: + for line in r: + yield line.rstrip('\n') + + +def create_dir_for_file(filename): + if parent := os.path.dirname(filename): + os.makedirs(parent, exist_ok=True) + + +def input_file(): + return click.Path(exists=True, file_okay=True, dir_okay=False, + writable=False, readable=True, resolve_path=True) + + +def input_dir(): + return click.Path(exists=True, file_okay=False, dir_okay=True, + writable=False, readable=True, resolve_path=True) + + +def output_file(): + return click.Path(exists=False, file_okay=True, dir_okay=False, + writable=True, readable=True, resolve_path=True) + + +def output_dir(): + return click.Path(exists=False, file_okay=False, dir_okay=True, + writable=True, readable=True, resolve_path=True) + + +class PasswordType(StrEnum): + 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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5baa7c0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "collectionparser" +version = "3.0.0" +description = "Command line tools to parse and query multi-billion password collections." +requires-python = ">=3.11" +dependencies = [ + "click>=8.1", + "tqdm>=4.66", +] + +[project.optional-dependencies] +test = ["pytest>=8"] + +[project.scripts] +collectionparser = "collectionparser.cli:cli" + +[tool.setuptools.packages.find] +include = ["collectionparser*"] diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 6855e47..0000000 --- a/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -tqdm -validate-email -click -cython \ No newline at end of file diff --git a/setup.py b/setup.py deleted file mode 100644 index ef707d0..0000000 --- a/setup.py +++ /dev/null @@ -1,26 +0,0 @@ -import sys - -from setuptools import setup, find_packages -from Cython.Build import cythonize - -if sys.version_info < (3, 6): - raise Exception('Must be using Python 3.6 or above.') - -# Install cython first. -setup(name='breach', - version='2.2', - author='Philippe Remy', - ext_modules=cythonize("breach/cparser.pyx", language_level="3"), - packages=find_packages(), - install_requires=[ - 'tqdm', - 'validate-email', - 'click' - ], - include_package_data=True, - entry_points={ - 'console_scripts': [ - 'breach=breach.cli:cli', - ], - } - ) diff --git a/tests/parser_test.py b/tests/parser_test.py index 4b27dbc..eaabe7b 100644 --- a/tests/parser_test.py +++ b/tests/parser_test.py @@ -1,16 +1,8 @@ import unittest from time import time -from breach.parser import LineParser -from breach.splitter import get_output_file -from breach.utils import alpha_num_lookup - -from breach.cparser import c_parseline - -p_fast = c_parseline -p_slow = LineParser.parseline - -p = p_fast +from collectionparser.parser import parseline as p +from collectionparser.splitter import get_output_file class ParserTest(unittest.TestCase): @@ -53,6 +45,10 @@ class ParserTest(unittest.TestCase): self.assertIsNone(p('ESSENTIALWAY')) self.assertIsNone(p('ESTARRON-ESPINOSA')) self.assertIsNone(p('ERLONMIRANDA:sha98ne')) + # No '@' and no '%40' -> cannot yield a pair, must stay None. + self.assertIsNone(p('no at sign at all')) + self.assertIsNone(p('plain text line:with:colons')) + self.assertIsNotNone(p('user%40domain.com:password')) # %40 still becomes '@' self.assertIsNotNone(p('hookturn54@hotmail.com:3318flindo2003@yahoo.ca:victory')) self.assertIsNotNone(p('mbssundaram@gmail.com;;shanmuga6')) self.assertIsNotNone(p('mbssundaram@gmail.com::shanmuga6')) @@ -109,7 +105,7 @@ class ParserTest(unittest.TestCase): def test_parse_2(self): t = time() - for i in range(20000): # 10,000 * 13 = 130,000 in 4.0 seconds. + for _ in range(20000): # 10,000 * 13 = 130,000 in 4.0 seconds. # => 32B in in 10 days. That's a lot. p('hello@gmail.com:123') p('hello@@gmail.com:123') @@ -128,13 +124,12 @@ class ParserTest(unittest.TestCase): print(time() - t) def test_parse_3(self): - alpha = alpha_num_lookup() - self.assertEqual('j/o/s', get_output_file(alpha, 'josemariacanyero@hotmail.com')) - self.assertEqual('c/e/o', get_output_file(alpha, 'ceo21w@gmail.com')) - self.assertEqual('x/symbols', get_output_file(alpha, 'x_fear_238@hotmail.fr')) - self.assertEqual('x/a/symbols', get_output_file(alpha, 'xa_fear_238@hotmail.fr')) - self.assertEqual('x/symbols', get_output_file(alpha, 'x@hotmail.com')) - self.assertEqual('k/e/v', get_output_file(alpha, 'kevmedic13@yahoo.com')) + self.assertEqual('j/o/s', get_output_file('josemariacanyero@hotmail.com')) + self.assertEqual('c/e/o', get_output_file('ceo21w@gmail.com')) + self.assertEqual('x/symbols', get_output_file('x_fear_238@hotmail.fr')) + self.assertEqual('x/a/symbols', get_output_file('xa_fear_238@hotmail.fr')) + self.assertEqual('x/symbols', get_output_file('x@hotmail.com')) + self.assertEqual('k/e/v', get_output_file('kevmedic13@yahoo.com')) if __name__ == '__main__': diff --git a/tests/query_test.py b/tests/query_test.py deleted file mode 100644 index dedca33..0000000 --- a/tests/query_test.py +++ /dev/null @@ -1,33 +0,0 @@ -import unittest - -from breach.query import mask_pass, mask_pass_justin_rule - - -class ParserTest(unittest.TestCase): - - def test_mask_password(self): - self.assertEqual('123456**654321', mask_pass('12345688654321', num_asterisks=2)) - self.assertEqual('123456***54321', mask_pass('12345688654321', num_asterisks=3)) - self.assertEqual('12345****54321', mask_pass('12345688654321', num_asterisks=4)) - self.assertEqual('12**********21', mask_pass('12345688654321', num_asterisks=10)) - self.assertEqual('1************1', mask_pass('12345688654321', num_asterisks=12)) - self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=14)) - self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=18)) - self.assertEqual('**************', mask_pass('12345688654321', num_asterisks=99)) - self.assertEqual('1234568*654321', mask_pass('12345688654321', num_asterisks=1)) - self.assertEqual('12345688654321', mask_pass('12345688654321', num_asterisks=0)) - - def test_mask_password_justin(self): - expected = 'il**********m' - initial = 'iloveicecream' - assert len(expected) == len(initial) - self.assertEqual(expected, mask_pass_justin_rule(initial, num_asterisks=-1)) - - expected = 'm****n' - initial = 'martin' - assert len(expected) == len(initial) - self.assertEqual(expected, mask_pass_justin_rule(initial, num_asterisks=-1)) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/utils_test.py b/tests/utils_test.py index d8dff02..bfe065b 100644 --- a/tests/utils_test.py +++ b/tests/utils_test.py @@ -1,6 +1,6 @@ import unittest -from breach.utils import RingDictionary +from collectionparser.utils import RingDictionary class UtilsTest(unittest.TestCase): @@ -8,27 +8,27 @@ class UtilsTest(unittest.TestCase): def test_ring_dictionary_1(self): rd = RingDictionary(max_num_keys=3) rd['a'] = 2 - self.assertListEqual(sorted(rd.keys()), ['a']) + self.assertListEqual(sorted(rd.d), ['a']) self.assertListEqual(sorted(rd.k), ['a']) rd['b'] = 2 - self.assertListEqual(sorted(rd.keys()), ['a', 'b']) + self.assertListEqual(sorted(rd.d), ['a', 'b']) self.assertListEqual(sorted(rd.k), ['a', 'b']) rd['c'] = 2 - self.assertListEqual(sorted(rd.keys()), ['a', 'b', 'c']) + self.assertListEqual(sorted(rd.d), ['a', 'b', 'c']) self.assertListEqual(sorted(rd.k), ['a', 'b', 'c']) rd['d'] = 2 - self.assertListEqual(sorted(rd.keys()), ['b', 'c', 'd']) + 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.keys()), ['c']) + self.assertListEqual(sorted(rd.d), ['c']) self.assertListEqual(sorted(rd.k), ['c']) rd['d'] = 3 - self.assertListEqual(sorted(rd.keys()), ['d']) + self.assertListEqual(sorted(rd.d), ['d']) self.assertListEqual(sorted(rd.k), ['d'])