diff --git a/hash_testing/new_testing/generate_input.py b/hash_testing/new_testing/generate_input.py new file mode 100755 index 0000000..756e601 --- /dev/null +++ b/hash_testing/new_testing/generate_input.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Generate unique words similar to a given word by applying random mutations.""" + +from __future__ import annotations + +import argparse +import random +import string +import sys +from collections.abc import Callable, Sequence +from pathlib import Path + +DEFAULT_ALPHABET = string.ascii_letters + string.digits + "_" + +OPERATION_ALIASES = { + "1": "replace", + "replace": "replace", + "2": "delete", + "delete": "delete", + "3": "add", + "add": "add", + "4": "swap", + "swap": "swap", + "5": "case", + "case": "case", + "6": "first", + "first": "first", + "7": "last", + "last": "last", +} + +OPERATION_HELP = """операция изменения: + 1, replace заменить случайный символ + 2, delete удалить случайный символ + 3, add добавить символ в случайную позицию + 4, swap переставить два соседних символа + 5, case сменить регистр случайного символа + 6, first изменить первый символ + 7, last изменить последний символ""" + + +class MutationError(ValueError): + """Raised when the selected mutation cannot be applied.""" + + +def different_character(current: str, alphabet: str, rng: random.Random) -> str: + choices = [character for character in alphabet if character != current] + if not choices: + raise MutationError("алфавит не содержит символа, отличного от заменяемого") + return rng.choice(choices) + + +def replace_character(word: str, alphabet: str, rng: random.Random) -> str: + if not word: + raise MutationError("нельзя заменить символ в пустом слове") + index = rng.randrange(len(word)) + replacement = different_character(word[index], alphabet, rng) + return word[:index] + replacement + word[index + 1 :] + + +def delete_character(word: str, _alphabet: str, rng: random.Random) -> str: + if not word: + raise MutationError("нельзя удалить символ из пустого слова") + index = rng.randrange(len(word)) + return word[:index] + word[index + 1 :] + + +def add_character(word: str, alphabet: str, rng: random.Random) -> str: + index = rng.randrange(len(word) + 1) + return word[:index] + rng.choice(alphabet) + word[index:] + + +def swap_adjacent(word: str, _alphabet: str, rng: random.Random) -> str: + indexes = [ + index for index in range(len(word) - 1) if word[index] != word[index + 1] + ] + if not indexes: + raise MutationError( + "для перестановки нужны хотя бы два соседних различных символа" + ) + index = rng.choice(indexes) + return word[:index] + word[index + 1] + word[index] + word[index + 2 :] + + +def change_case(word: str, _alphabet: str, rng: random.Random) -> str: + indexes = [] + replacements: dict[int, str] = {} + for index, character in enumerate(word): + swapped = character.swapcase() + if swapped != character and len(swapped) == 1: + indexes.append(index) + replacements[index] = swapped + if not indexes: + raise MutationError("в слове нет символов, у которых можно сменить регистр") + index = rng.choice(indexes) + return word[:index] + replacements[index] + word[index + 1 :] + + +def change_first(word: str, alphabet: str, rng: random.Random) -> str: + if not word: + raise MutationError("нельзя изменить первый символ пустого слова") + replacement = different_character(word[0], alphabet, rng) + return replacement + word[1:] + + +def change_last(word: str, alphabet: str, rng: random.Random) -> str: + if not word: + raise MutationError("нельзя изменить последний символ пустого слова") + replacement = different_character(word[-1], alphabet, rng) + return word[:-1] + replacement + + +Mutation = Callable[[str, str, random.Random], str] +MUTATIONS: dict[str, Mutation] = { + "replace": replace_character, + "delete": delete_character, + "add": add_character, + "swap": swap_adjacent, + "case": change_case, + "first": change_first, + "last": change_last, +} + + +def parse_operation(value: str) -> str: + try: + return OPERATION_ALIASES[value.lower()] + except KeyError as error: + valid = ", ".join(OPERATION_ALIASES) + raise argparse.ArgumentTypeError( + f"неизвестная операция {value!r}; допустимы: {valid}" + ) from error + + +def generate_words( + source: str, + count: int, + operation: str, + operation_count: int, + alphabet: str = DEFAULT_ALPHABET, + seed: int | None = None, + max_attempts: int | None = None, +) -> list[str]: + """Generate up to ``count`` unique mutations made by exactly N operations.""" + if count < 1: + raise ValueError("количество выходных слов должно быть положительным") + if operation_count < 1: + raise ValueError("количество операций должно быть положительным") + if not alphabet: + raise ValueError("алфавит не должен быть пустым") + + mutation = MUTATIONS[operation] + rng = random.Random(seed) + attempt_limit = max_attempts or max(10_000, count * 1_000) + words: list[str] = [] + seen: set[str] = set() + + for _attempt in range(attempt_limit): + candidate = source + try: + for _ in range(operation_count): + candidate = mutation(candidate, alphabet, rng) + except MutationError: + continue + + if candidate != source and candidate not in seen: + seen.add(candidate) + words.append(candidate) + if len(words) == count: + return words + + return words + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Создаёт список уникальных слов, похожих на исходное. " + "Каждое слово получается независимо от исходного ровно заданным " + "числом случайных операций." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + f"{OPERATION_HELP}\n\n" + "Пример:\n" + " python3 nexus.py example -o swap -n 20 -k 2 --seed 42" + ), + ) + parser.add_argument("word", help="исходное слово") + parser.add_argument( + "-o", "--operation", required=True, type=parse_operation, help=OPERATION_HELP + ) + parser.add_argument( + "-n", + "--count", + type=int, + default=10, + help="верхняя граница числа уникальных выходных слов (по умолчанию: 10)", + ) + parser.add_argument( + "-k", + "--operations", + type=int, + default=1, + help="число операций над каждым словом (по умолчанию: 1)", + ) + parser.add_argument( + "--alphabet", + default=DEFAULT_ALPHABET, + help=("символы для добавления и замены " f"(по умолчанию: {DEFAULT_ALPHABET})"), + ) + parser.add_argument( + "--seed", type=int, help="seed генератора для воспроизводимого результата" + ) + parser.add_argument( + "--max-attempts", + type=int, + help="предельное число попыток собрать уникальные слова", + ) + parser.add_argument( + "--output", + type=Path, + help="записать слова в файл вместо стандартного вывода", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.max_attempts is not None and args.max_attempts < 1: + parser.error("--max-attempts должен быть положительным") + + try: + words = generate_words( + source=args.word, + count=args.count, + operation=args.operation, + operation_count=args.operations, + alphabet=args.alphabet, + seed=args.seed, + max_attempts=args.max_attempts, + ) + except (MutationError, ValueError) as error: + parser.error(str(error)) + + if len(words) < args.count: + print( + f"warning: operation={args.operation}: generated {len(words)} " + f"of at most {args.count} unique words", + file=sys.stderr, + ) + + output = "".join(f"{word}\n" for word in words) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(output, encoding="utf-8") + else: + sys.stdout.write(output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash new file mode 100755 index 0000000..81868a8 Binary files /dev/null and b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash differ diff --git a/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash.c b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash.c new file mode 100644 index 0000000..2856005 --- /dev/null +++ b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/bin_hash.c @@ -0,0 +1,109 @@ +#include +#include +#include +#include +#include + +#define JOAAT_SEED UINT32_C(0x9e3779b9) + +static uint32_t jenkins_oaat(const unsigned char *data, size_t length) +{ + uint32_t hash = JOAAT_SEED; + + for (size_t index = 0; index < length; ++index) { + hash += data[index]; + hash += hash << 10; + hash ^= hash >> 6; + } + + hash += hash << 3; + hash ^= hash >> 11; + hash += hash << 15; + return hash; +} + +static int is_ascii_trailing_space(unsigned char character) +{ + return character == ' ' || character == '\t' || character == '\n' || + character == '\r' || character == '\v' || character == '\f'; +} + +static int read_stdin(unsigned char **data, size_t *length) +{ + size_t capacity = 256; + unsigned char *buffer = malloc(capacity); + + if (buffer == NULL) { + return -1; + } + + *length = 0; + for (;;) { + size_t available = capacity - *length; + size_t bytes_read = fread(buffer + *length, 1, available, stdin); + *length += bytes_read; + + if (bytes_read < available) { + if (ferror(stdin)) { + free(buffer); + return -1; + } + break; + } + + if (capacity > SIZE_MAX / 2) { + free(buffer); + return -1; + } + capacity *= 2; + + unsigned char *larger_buffer = realloc(buffer, capacity); + if (larger_buffer == NULL) { + free(buffer); + return -1; + } + buffer = larger_buffer; + } + + *data = buffer; + return 0; +} + +int main(int argc, char **argv) +{ + const unsigned char *word; + unsigned char *stdin_buffer = NULL; + size_t length; + + if (argc > 2) { + fprintf(stderr, "usage: %s [ASCII_WORD]\n", argv[0]); + return EXIT_FAILURE; + } + + if (argc == 2) { + word = (const unsigned char *)argv[1]; + length = strlen(argv[1]); + } else { + if (read_stdin(&stdin_buffer, &length) != 0) { + fprintf(stderr, "failed to read input\n"); + return EXIT_FAILURE; + } + word = stdin_buffer; + } + + while (length > 0 && is_ascii_trailing_space(word[length - 1])) { + --length; + } + + for (size_t index = 0; index < length; ++index) { + if (word[index] > 0x7f) { + fprintf(stderr, "input must contain ASCII characters only\n"); + free(stdin_buffer); + return EXIT_FAILURE; + } + } + + printf("%08" PRIx32 "\n", jenkins_oaat(word, length)); + free(stdin_buffer); + return EXIT_SUCCESS; +} \ No newline at end of file diff --git a/hash_testing/new_testing/hash_funcs/jenkinsOAAT/hash.c b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/hash.c new file mode 100644 index 0000000..e0bbbd5 --- /dev/null +++ b/hash_testing/new_testing/hash_funcs/jenkinsOAAT/hash.c @@ -0,0 +1,13 @@ +unsigned int hash(const char* str) { + unsigned int hash = 0x9e3779b9; + const unsigned char* p = (const unsigned char*)str; + while (*p) { + hash += *p++; + hash += (hash << 10); + hash ^= (hash >> 6); + } + hash += (hash << 3); + hash ^= (hash >> 11); + hash += (hash << 15); + return hash; +} diff --git a/hash_testing/new_testing/plot_probability_map.py b/hash_testing/new_testing/plot_probability_map.py new file mode 100755 index 0000000..62708df --- /dev/null +++ b/hash_testing/new_testing/plot_probability_map.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +"""Render avalanche bit-probability CSV files as two PNG bar charts.""" + +from __future__ import annotations + +import argparse +import csv +import math +from dataclasses import dataclass +from pathlib import Path +from statistics import fmean +from typing import Sequence + +from PIL import Image, ImageDraw, ImageFont + +BACKGROUND = "#f7f8fa" +PANEL = "#ffffff" +GRID = "#d9dee7" +TEXT = "#172033" +MUTED = "#637083" +REFERENCE = "#d24b4b" +COLORS = ( + "#377eb8", + "#4daf4a", + "#984ea3", + "#ff7f00", + "#e41a1c", + "#00a6a6", + "#a65628", +) + + +@dataclass(frozen=True) +class ProbabilityRow: + operation: str + pairs: int + probabilities: list[float] + + +@dataclass(frozen=True) +class ChartScale: + minimum: float + maximum: float + ticks: tuple[float, ...] + + +def nice_step(value: float) -> float: + exponent = math.floor(math.log10(value)) + fraction = value / 10**exponent + nice_fraction = min((1.0, 2.0, 2.5, 5.0, 10.0), key=lambda item: abs(item - fraction)) + return nice_fraction * 10**exponent + + +def make_scale( + values: Sequence[float], + *, + hard_limits: tuple[float, float], + reference: float | None = None, +) -> ChartScale: + """Build a padded shared scale constrained to hard limits.""" + hard_minimum, hard_maximum = hard_limits + finite_values = [value for value in values if math.isfinite(value)] + if reference is not None: + finite_values.append(reference) + if not finite_values: + finite_values = [hard_minimum, hard_maximum] + + minimum = min(finite_values) + maximum = max(finite_values) + if minimum == maximum: + expansion = (hard_maximum - hard_minimum) * 0.1 + minimum -= expansion / 2 + maximum += expansion / 2 + + span = maximum - minimum + padded_minimum = max(hard_minimum, minimum - span * 0.1) + padded_maximum = min(hard_maximum, maximum + span * 0.1) + step = nice_step(max((padded_maximum - padded_minimum) / 5, 1e-12)) + scaled_minimum = max(hard_minimum, math.floor(padded_minimum / step) * step) + scaled_maximum = min(hard_maximum, math.ceil(padded_maximum / step) * step) + if scaled_minimum == scaled_maximum: + scaled_minimum, scaled_maximum = hard_minimum, hard_maximum + + tick_count = round((scaled_maximum - scaled_minimum) / step) + ticks = [scaled_minimum + index * step for index in range(tick_count + 1)] + if reference is not None and scaled_minimum <= reference <= scaled_maximum: + ticks.append(reference) + normalized_ticks = tuple( + sorted({round(value, 12) for value in ticks if scaled_minimum <= value <= scaled_maximum}) + ) + return ChartScale(scaled_minimum, scaled_maximum, normalized_ticks) + + +def load_font(size: int, bold: bool = False) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: + names = ( + "DejaVuSans-Bold.ttf" if bold else "DejaVuSans.ttf", + "/usr/share/fonts/TTF/DejaVuSans-Bold.ttf" + if bold + else "/usr/share/fonts/TTF/DejaVuSans.ttf", + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" + if bold + else "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + ) + for name in names: + try: + return ImageFont.truetype(name, size) + except OSError: + continue + return ImageFont.load_default() + + +def read_probability_map(path: Path) -> list[ProbabilityRow]: + """Read operation rows and numerically ordered bit columns from a CSV file.""" + try: + stream = path.open(encoding="utf-8", newline="") + except OSError as error: + raise ValueError(f"не удалось открыть {path}: {error}") from error + + with stream: + reader = csv.DictReader(stream) + fields = reader.fieldnames + if not fields or "operation" not in fields or "pairs" not in fields: + raise ValueError("CSV должен содержать колонки operation и pairs") + + bit_fields: list[tuple[int, str]] = [] + for field in fields: + if not field.startswith("bit_"): + continue + try: + bit_fields.append((int(field.removeprefix("bit_")), field)) + except ValueError as error: + raise ValueError(f"некорректная битовая колонка: {field}") from error + bit_fields.sort() + if not bit_fields: + raise ValueError("CSV не содержит колонок bit_N") + expected_bits = list(range(len(bit_fields))) + actual_bits = [bit for bit, _field in bit_fields] + if actual_bits != expected_bits: + raise ValueError("битовые колонки должны непрерывно идти от bit_0") + + rows: list[ProbabilityRow] = [] + for line_number, row in enumerate(reader, start=2): + operation = (row.get("operation") or "").strip() + if not operation: + raise ValueError(f"строка {line_number}: пустая операция") + try: + pairs = int(row["pairs"] or "") + except (TypeError, ValueError) as error: + raise ValueError( + f"строка {line_number}: некорректное число пар" + ) from error + if pairs < 0: + raise ValueError(f"строка {line_number}: число пар меньше нуля") + + probabilities: list[float] = [] + for _bit, field in bit_fields: + try: + value = float(row[field] or "") + except (TypeError, ValueError) as error: + raise ValueError( + f"строка {line_number}: некорректное значение {field}" + ) from error + if not math.isnan(value) and not 0.0 <= value <= 1.0: + raise ValueError( + f"строка {line_number}: {field} должен быть от 0 до 1" + ) + probabilities.append(value) + rows.append(ProbabilityRow(operation, pairs, probabilities)) + + if not rows: + raise ValueError("CSV не содержит строк с операциями") + return rows + + +def mean_absolute_deviation(probabilities: Sequence[float]) -> float: + """Return the mean |p - 0.5| over finite bit probabilities.""" + deviations = [ + abs(probability - 0.5) + for probability in probabilities + if math.isfinite(probability) + ] + return fmean(deviations) if deviations else float("nan") + + +def text_width( + draw: ImageDraw.ImageDraw, + text: str, + font: ImageFont.FreeTypeFont | ImageFont.ImageFont, +) -> int: + box = draw.textbbox((0, 0), text, font=font) + return round(box[2] - box[0]) + + +def draw_centered_text( + draw: ImageDraw.ImageDraw, + center_x: float, + y: float, + text: str, + font: ImageFont.FreeTypeFont | ImageFont.ImageFont, + fill: str = TEXT, +) -> None: + draw.text( + (center_x - text_width(draw, text, font) / 2, y), + text, + font=font, + fill=fill, + ) + + +def draw_bit_panel( + draw: ImageDraw.ImageDraw, + bounds: tuple[int, int, int, int], + row: ProbabilityRow, + color: str, + scale: ChartScale, +) -> None: + left, top, right, bottom = bounds + title_font = load_font(22, bold=True) + label_font = load_font(14) + tick_font = load_font(12) + + draw.rounded_rectangle(bounds, radius=12, fill=PANEL, outline=GRID, width=1) + draw.text((left + 18, top + 13), row.operation, font=title_font, fill=TEXT) + pairs_text = f"pairs: {row.pairs}" + draw.text( + (right - 18 - text_width(draw, pairs_text, label_font), top + 17), + pairs_text, + font=label_font, + fill=MUTED, + ) + + plot_left = left + 54 + plot_right = right - 18 + plot_top = top + 55 + plot_bottom = bottom - 42 + plot_height = plot_bottom - plot_top + + scale_span = scale.maximum - scale.minimum + for probability in scale.ticks: + y = round(plot_bottom - (probability - scale.minimum) / scale_span * plot_height) + line_color = REFERENCE if probability == 0.5 else GRID + line_width = 2 if probability == 0.5 else 1 + draw.line((plot_left, y, plot_right, y), fill=line_color, width=line_width) + label = f"{probability:.3g}" + draw.text( + (plot_left - 8 - text_width(draw, label, tick_font), y - 7), + label, + font=tick_font, + fill=MUTED, + ) + + bit_count = len(row.probabilities) + slot_width = (plot_right - plot_left) / bit_count + bar_width = max(1, int(slot_width * 0.72)) + for bit, probability in enumerate(row.probabilities): + if not math.isfinite(probability): + continue + center = plot_left + (bit + 0.5) * slot_width + x0 = round(center - bar_width / 2) + x1 = round(center + bar_width / 2) + y = round( + plot_bottom + - (probability - scale.minimum) / scale_span * plot_height + ) + draw.rectangle((x0, y, x1, plot_bottom), fill=color) + + tick_step = max(1, math.ceil(bit_count / 16)) + for bit in range(0, bit_count, tick_step): + center = plot_left + (bit + 0.5) * slot_width + label = str(bit) + draw.text( + (center - text_width(draw, label, tick_font) / 2, plot_bottom + 7), + label, + font=tick_font, + fill=MUTED, + ) + draw_centered_text( + draw, + (plot_left + plot_right) / 2, + bottom - 21, + "output bit (0 = LSB)", + tick_font, + MUTED, + ) + + +def render_bit_probabilities( + rows: Sequence[ProbabilityRow], output: Path, title: str +) -> None: + columns = 2 if len(rows) > 1 else 1 + panel_width = 760 + panel_height = 330 + gap = 18 + margin = 24 + title_height = 70 + row_count = math.ceil(len(rows) / columns) + width = margin * 2 + columns * panel_width + (columns - 1) * gap + height = title_height + margin + row_count * panel_height + (row_count - 1) * gap + + image = Image.new("RGB", (width, height), BACKGROUND) + draw = ImageDraw.Draw(image) + scale = make_scale( + [probability for row in rows for probability in row.probabilities], + reference=0.5, + hard_limits=(0.0, 1.0), + ) + draw_centered_text(draw, width / 2, 18, title, load_font(30, bold=True)) + draw_centered_text( + draw, + width / 2, + 52, + f"Shared scale {scale.minimum:.3g}–{scale.maximum:.3g}; red line = ideal p=0.5", + load_font(14), + MUTED, + ) + + for index, row in enumerate(rows): + column = index % columns + grid_row = index // columns + left = margin + column * (panel_width + gap) + top = title_height + grid_row * (panel_height + gap) + draw_bit_panel( + draw, + (left, top, left + panel_width, top + panel_height), + row, + COLORS[index % len(COLORS)], + scale, + ) + + output.parent.mkdir(parents=True, exist_ok=True) + image.save(output, "PNG", optimize=True) + + +def render_mean_deviations( + rows: Sequence[ProbabilityRow], output: Path, title: str +) -> None: + width, height = 1200, 720 + image = Image.new("RGB", (width, height), BACKGROUND) + draw = ImageDraw.Draw(image) + draw_centered_text(draw, width / 2, 22, title, load_font(30, bold=True)) + draw_centered_text( + draw, + width / 2, + 58, + "Mean absolute deviation from ideal avalanche probability: mean(|p - 0.5|)", + load_font(15), + MUTED, + ) + + plot_left, plot_right = 90, width - 40 + plot_top, plot_bottom = 110, height - 120 + plot_height = plot_bottom - plot_top + deviations = [mean_absolute_deviation(row.probabilities) for row in rows] + scale = make_scale(deviations, hard_limits=(0.0, 0.5)) + scale_span = scale.maximum - scale.minimum + tick_font = load_font(13) + label_font = load_font(15) + value_font = load_font(14, bold=True) + + for value in scale.ticks: + y = round(plot_bottom - (value - scale.minimum) / scale_span * plot_height) + draw.line((plot_left, y, plot_right, y), fill=GRID, width=1) + label = f"{value:.3g}" + draw.text( + (plot_left - 10 - text_width(draw, label, tick_font), y - 7), + label, + font=tick_font, + fill=MUTED, + ) + + slot_width = (plot_right - plot_left) / len(rows) + bar_width = min(105, max(20, int(slot_width * 0.62))) + for index, (row, deviation) in enumerate(zip(rows, deviations, strict=True)): + center = plot_left + (index + 0.5) * slot_width + x0 = round(center - bar_width / 2) + x1 = round(center + bar_width / 2) + if math.isfinite(deviation): + y = round( + plot_bottom + - (deviation - scale.minimum) / scale_span * plot_height + ) + draw.rectangle((x0, y, x1, plot_bottom), fill=COLORS[index % len(COLORS)]) + value_label = f"{deviation:.4f}" + else: + y = plot_bottom + value_label = "n/a" + draw_centered_text(draw, center, max(plot_top, y - 22), value_label, value_font) + draw_centered_text(draw, center, plot_bottom + 12, row.operation, label_font) + draw_centered_text( + draw, center, plot_bottom + 36, f"pairs: {row.pairs}", tick_font, MUTED + ) + + draw.line((plot_left, plot_top, plot_left, plot_bottom), fill=TEXT, width=2) + draw.line((plot_left, plot_bottom, plot_right, plot_bottom), fill=TEXT, width=2) + output.parent.mkdir(parents=True, exist_ok=True) + image.save(output, "PNG", optimize=True) + + +def generate_plots(source: Path, output_directory: Path | None = None) -> tuple[Path, Path]: + rows = read_probability_map(source) + destination = output_directory or source.parent + bit_output = destination / f"{source.stem}_bits.png" + deviation_output = destination / f"{source.stem}_deviation.png" + display_name = source.stem + + render_bit_probabilities(rows, bit_output, f"{display_name}: per-bit avalanche map") + render_mean_deviations( + rows, + deviation_output, + f"{display_name}: deviation from p=0.5", + ) + return bit_output, deviation_output + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Создаёт две PNG-гистограммы из CSV вероятностной карты: " + "вероятности по битам для каждой операции и среднее |p-0.5|." + ) + ) + parser.add_argument("csv_file", type=Path, help="CSV из probability_map.py") + parser.add_argument( + "-o", + "--output-dir", + type=Path, + help="каталог PNG (по умолчанию каталог исходного CSV)", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + outputs = generate_plots(args.csv_file, args.output_dir) + except ValueError as error: + raise SystemExit(f"error: {error}") from error + + for output in outputs: + print(f"wrote {output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/hash_testing/new_testing/probability_map.py b/hash_testing/new_testing/probability_map.py new file mode 100755 index 0000000..d9bfd9c --- /dev/null +++ b/hash_testing/new_testing/probability_map.py @@ -0,0 +1,364 @@ +#!/usr/bin/env python3 +"""Build bit-probability maps for hash functions and word mutation types. + +For every hash listed in HASHES, the script compares each source word hash with +hashes of generated similar words. A table cell contains the probability that +the corresponding output bit changed (XOR with its source word hash). +""" + +from __future__ import annotations + +import argparse +import csv +import re +import stat +import subprocess +import sys +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import TextIO + +from generate_input import ( + DEFAULT_ALPHABET, + MUTATIONS, + MutationError, + generate_words, + parse_operation, +) + +ROOT = Path(__file__).resolve().parent +HASH_FUNCS_DIR = ROOT / "hash_funcs" +DEFAULT_OUTPUT_DIR = ROOT / "probability_maps" + +# Add directory names from hash_funcs here to include more implementations. +HASHES = [ + "jenkinsOAAT", +] + +HEX_HASH = re.compile(r"(?:0[xX])?([0-9a-fA-F]+)") +ProbabilityRow = tuple[int, list[float]] + + +class HashToolError(RuntimeError): + """Raised when a hash executable cannot be prepared or invoked.""" + + +def prepare_hash(hash_name: str, hash_funcs_dir: Path = HASH_FUNCS_DIR) -> Path: + """Return an executable hash tool, building or preparing it if necessary.""" + if not hash_name or Path(hash_name).name != hash_name: + raise HashToolError(f"некорректное имя хэша: {hash_name!r}") + + hash_directory = hash_funcs_dir / hash_name + if not hash_directory.is_dir(): + raise HashToolError(f"не найдена папка хэша: {hash_directory}") + + binary = hash_directory / "bin_hash" + if binary.is_file(): + binary.chmod(binary.stat().st_mode | stat.S_IXUSR) + return binary + + sources = ( + (hash_directory / "bin_hash.c", "cc"), + (hash_directory / "bin_hash.cpp", "c++"), + (hash_directory / "bin_hash.cc", "c++"), + (hash_directory / "bin_hash.cxx", "c++"), + ) + for source, compiler in sources: + if not source.is_file(): + continue + command = [ + compiler, + "-O2", + "-Wall", + "-Wextra", + "-Wpedantic", + "-Werror", + str(source), + "-o", + str(binary), + ] + if compiler == "cc": + command[1:1] = ["-std=c11"] + else: + command[1:1] = ["-std=c++17"] + result = subprocess.run(command, text=True, capture_output=True) + if result.returncode != 0: + details = result.stderr.strip() or result.stdout.strip() + raise HashToolError( + f"не удалось скомпилировать {source}: {details}" + ) + return binary + + python_source = hash_directory / "bin_hash.py" + if python_source.is_file(): + content = python_source.read_text(encoding="utf-8") + if not content.startswith("#!"): + python_source.write_text( + "#!/usr/bin/env python3\n" + content, + encoding="utf-8", + ) + python_source.chmod( + python_source.stat().st_mode + | stat.S_IXUSR + | stat.S_IXGRP + | stat.S_IXOTH + ) + return python_source + + raise HashToolError( + f"для {hash_name} не найден bin_hash, bin_hash.c/cpp/cc/cxx или bin_hash.py" + ) + + +def run_hash(executable: Path, word: str) -> tuple[int, int]: + """Run a hash tool and return its integer value and explicit output width.""" + try: + result = subprocess.run( + [str(executable), word], + text=True, + capture_output=True, + check=False, + ) + except OSError as error: + raise HashToolError(f"не удалось запустить {executable}: {error}") from error + + if result.returncode != 0: + details = result.stderr.strip() or result.stdout.strip() + raise HashToolError( + f"{executable} завершился с кодом {result.returncode}: {details}" + ) + + output = result.stdout.strip() + match = HEX_HASH.fullmatch(output) + if match is None: + raise HashToolError( + f"{executable} вернул не шестнадцатеричный хэш: {output!r}" + ) + + digits = match.group(1) + return int(digits, 16), len(digits) * 4 + + +def bit_probabilities( + source_hash: int, changed_hashes: Sequence[int], bits: int +) -> list[float]: + """Calculate per-bit change probabilities, ordered from LSB to MSB.""" + if bits < 1: + raise ValueError("число бит должно быть положительным") + if not changed_hashes: + raise ValueError("список изменённых хэшей не должен быть пустым") + + changed_counts = [0] * bits + for changed_hash in changed_hashes: + difference = source_hash ^ changed_hash + for bit in range(bits): + changed_counts[bit] += (difference >> bit) & 1 + + sample_count = len(changed_hashes) + return [count / sample_count for count in changed_counts] + + +def write_csv_table(stream: TextIO, rows: Mapping[str, ProbabilityRow]) -> None: + """Write one operation-by-bit probability table as CSV.""" + if not rows: + raise ValueError("таблица вероятностей не должна быть пустой") + + widths = {len(probabilities) for _pair_count, probabilities in rows.values()} + if len(widths) != 1: + raise ValueError("все строки таблицы должны иметь одинаковое число бит") + bits = widths.pop() + + writer = csv.writer(stream, lineterminator="\n") + writer.writerow( + ["operation", "pairs", *(f"bit_{bit}" for bit in range(bits))] + ) + for operation, (pair_count, probabilities) in rows.items(): + writer.writerow( + [ + operation, + pair_count, + *(f"{probability:.6f}" for probability in probabilities), + ] + ) + + +def build_probability_table( + executable: Path, + sources: Sequence[str], + operations: Sequence[str], + count: int, + operation_count: int, + alphabet: str, + seed: int | None, + max_attempts: int | None, +) -> dict[str, ProbabilityRow]: + """Aggregate avalanche probabilities across all source words.""" + if not sources: + raise ValueError("нужно указать хотя бы одно исходное слово") + + source_hashes: list[int] = [] + bits: int | None = None + for source in sources: + source_hash, source_bits = run_hash(executable, source) + if bits is None: + bits = source_bits + elif source_bits != bits: + raise HashToolError( + f"{executable} вернул хэши разной ширины: " + f"{bits} и {source_bits} бит" + ) + source_hashes.append(source_hash) + + assert bits is not None + table: dict[str, ProbabilityRow] = {} + + for operation in operations: + differences: list[int] = [] + for source, source_hash in zip(sources, source_hashes, strict=True): + words = generate_words( + source=source, + count=count, + operation=operation, + operation_count=operation_count, + alphabet=alphabet, + seed=seed, + max_attempts=max_attempts, + ) + if len(words) < count: + print( + f"warning: operation={operation} word={source!r}: " + f"generated {len(words)} of at most {count} unique words", + file=sys.stderr, + ) + for word in words: + changed_hash, changed_bits = run_hash(executable, word) + if changed_bits != bits: + raise HashToolError( + f"{executable} вернул хэши разной ширины: " + f"{bits} и {changed_bits} бит" + ) + differences.append(source_hash ^ changed_hash) + + if differences: + table[operation] = ( + len(differences), + bit_probabilities(0, differences, bits), + ) + else: + table[operation] = (0, [float("nan")] * bits) + + return table + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Строит для каждого хэша CSV-таблицу вероятностей изменения " + "выходных битов. bit_0 — младший бит." + ) + ) + parser.add_argument( + "words", + nargs="+", + help="одно или несколько исходных ASCII-слов", + ) + parser.add_argument( + "-o", + "--operation", + action="append", + type=parse_operation, + help=( + "тип изменения (номер или имя как в generate_input.py); " + "можно повторять, по умолчанию используются все типы" + ), + ) + parser.add_argument( + "-n", + "--count", + type=int, + default=10, + help=( + "верхняя граница числа уникальных изменённых слов для каждого " + "исходного слова и типа (по умолчанию: 10)" + ), + ) + parser.add_argument( + "-k", + "--operations", + type=int, + default=1, + help="число операций над каждым словом (по умолчанию: 1)", + ) + parser.add_argument( + "--alphabet", + default=DEFAULT_ALPHABET, + help="алфавит для добавления и замены", + ) + parser.add_argument( + "--seed", type=int, help="seed генератора для воспроизводимого результата" + ) + parser.add_argument( + "--max-attempts", + type=int, + help="предельное число попыток собрать уникальные слова", + ) + parser.add_argument( + "--output", + type=Path, + default=DEFAULT_OUTPUT_DIR, + help=( + "каталог для CSV-таблиц " + f"(по умолчанию: {DEFAULT_OUTPUT_DIR})" + ), + ) + parser.add_argument( + "--hash", + dest="hashes", + action="append", + help="проверить только указанный хэш; можно повторять (по умолчанию HASHES)", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.count < 1: + parser.error("--count должен быть положительным") + if args.operations < 1: + parser.error("--operations должен быть положительным") + if args.max_attempts is not None and args.max_attempts < 1: + parser.error("--max-attempts должен быть положительным") + + operations = list(dict.fromkeys(args.operation or MUTATIONS.keys())) + hashes = list(dict.fromkeys(args.hashes or HASHES)) + if not hashes: + parser.error("массив HASHES не должен быть пустым") + + args.output.mkdir(parents=True, exist_ok=True) + try: + for hash_name in hashes: + executable = prepare_hash(hash_name) + table = build_probability_table( + executable=executable, + sources=args.words, + operations=operations, + count=args.count, + operation_count=args.operations, + alphabet=args.alphabet, + seed=args.seed, + max_attempts=args.max_attempts, + ) + output_path = args.output / f"{hash_name}.csv" + with output_path.open("w", encoding="utf-8", newline="") as stream: + write_csv_table(stream, table) + print(f"wrote {output_path}") + except (HashToolError, MutationError, ValueError) as error: + parser.error(str(error)) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/hash_testing/new_testing/probability_maps/jenkinsOAAT.csv b/hash_testing/new_testing/probability_maps/jenkinsOAAT.csv new file mode 100644 index 0000000..8d768d0 --- /dev/null +++ b/hash_testing/new_testing/probability_maps/jenkinsOAAT.csv @@ -0,0 +1,5 @@ +operation,pairs,bit_0,bit_1,bit_2,bit_3,bit_4,bit_5,bit_6,bit_7,bit_8,bit_9,bit_10,bit_11,bit_12,bit_13,bit_14,bit_15,bit_16,bit_17,bit_18,bit_19,bit_20,bit_21,bit_22,bit_23,bit_24,bit_25,bit_26,bit_27,bit_28,bit_29,bit_30,bit_31 +replace,1922,0.486472,0.498959,0.483351,0.501561,0.503122,0.495317,0.523413,0.466701,0.520812,0.496878,0.513007,0.503122,0.490114,0.510926,0.482310,0.494277,0.511446,0.505723,0.498959,0.501561,0.512487,0.509365,0.489594,0.514048,0.495317,0.519771,0.505203,0.515609,0.517690,0.489074,0.498959,0.508845 +add,2111,0.506869,0.505448,0.502132,0.502132,0.496921,0.502132,0.505921,0.505921,0.494552,0.507342,0.493605,0.502605,0.492658,0.501658,0.495026,0.508764,0.496921,0.486973,0.507816,0.518238,0.513501,0.506395,0.513974,0.488394,0.504500,0.482236,0.510185,0.520133,0.500711,0.504027,0.520133,0.497868 +first,186,0.451613,0.435484,0.440860,0.462366,0.521505,0.494624,0.510753,0.435484,0.462366,0.483871,0.505376,0.521505,0.451613,0.532258,0.462366,0.505376,0.473118,0.505376,0.446237,0.473118,0.489247,0.537634,0.440860,0.521505,0.537634,0.500000,0.548387,0.575269,0.575269,0.494624,0.462366,0.505376 +last,186,0.516129,0.500000,0.483871,0.510753,0.510753,0.483871,0.478495,0.521505,0.569892,0.500000,0.478495,0.494624,0.462366,0.521505,0.510753,0.500000,0.569892,0.494624,0.505376,0.510753,0.521505,0.478495,0.478495,0.543011,0.500000,0.548387,0.532258,0.451613,0.500000,0.500000,0.516129,0.521505 diff --git a/hash_testing/new_testing/probability_maps/jenkinsOAAT_bits.png b/hash_testing/new_testing/probability_maps/jenkinsOAAT_bits.png new file mode 100644 index 0000000..e940149 Binary files /dev/null and b/hash_testing/new_testing/probability_maps/jenkinsOAAT_bits.png differ diff --git a/hash_testing/new_testing/probability_maps/jenkinsOAAT_deviation.png b/hash_testing/new_testing/probability_maps/jenkinsOAAT_deviation.png new file mode 100644 index 0000000..1da87a7 Binary files /dev/null and b/hash_testing/new_testing/probability_maps/jenkinsOAAT_deviation.png differ diff --git a/hash_testing/new_testing/test_plot_probability_map.py b/hash_testing/new_testing/test_plot_probability_map.py new file mode 100644 index 0000000..eac1cbf --- /dev/null +++ b/hash_testing/new_testing/test_plot_probability_map.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Behavior tests for plot_probability_map.py.""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from PIL import Image + +import plot_probability_map + + +class ProbabilityMapPlotTests(unittest.TestCase): + def test_reads_bit_columns_and_ignores_operation_and_pairs(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "hash.csv" + path.write_text( + "operation,pairs,bit_0,bit_1\n" + "replace,4,0.25,0.75\n" + "delete,2,0.5,nan\n", + encoding="utf-8", + ) + + rows = plot_probability_map.read_probability_map(path) + + self.assertEqual(rows[0].operation, "replace") + self.assertEqual(rows[0].pairs, 4) + self.assertEqual(rows[0].probabilities, [0.25, 0.75]) + self.assertEqual(rows[1].operation, "delete") + self.assertEqual(rows[1].pairs, 2) + self.assertEqual(len(rows[1].probabilities), 2) + + def test_mean_absolute_deviation_from_half_ignores_nan(self) -> None: + result = plot_probability_map.mean_absolute_deviation( + [0.25, 0.5, 0.75, float("nan")] + ) + + self.assertEqual(result, 1 / 6) + + def test_shared_scale_zooms_to_all_values_and_reference(self) -> None: + scale = plot_probability_map.make_scale( + [0.45, 0.48, 0.52, 0.55], + reference=0.5, + hard_limits=(0.0, 1.0), + ) + + self.assertGreater(scale.minimum, 0.0) + self.assertLess(scale.maximum, 1.0) + self.assertLessEqual(scale.minimum, 0.45) + self.assertGreaterEqual(scale.maximum, 0.55) + self.assertIn(0.5, scale.ticks) + + def test_deviation_scale_uses_data_range_instead_of_fixed_half(self) -> None: + scale = plot_probability_map.make_scale( + [0.05, 0.08], hard_limits=(0.0, 0.5) + ) + + self.assertGreater(scale.minimum, 0.0) + self.assertLess(scale.maximum, 0.5) + self.assertLessEqual(scale.minimum, 0.05) + self.assertGreaterEqual(scale.maximum, 0.08) + + def test_generates_two_nonempty_png_files(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + source = root / "sample.csv" + source.write_text( + "operation,pairs,bit_0,bit_1,bit_2,bit_3\n" + "replace,4,0.25,0.50,0.75,1.0\n" + "delete,2,0.10,0.20,0.30,0.40\n", + encoding="utf-8", + ) + + outputs = plot_probability_map.generate_plots(source, root / "plots") + + self.assertEqual(len(outputs), 2) + for output in outputs: + self.assertTrue(output.is_file()) + with Image.open(output) as image: + self.assertEqual(image.format, "PNG") + self.assertGreater(image.width, 300) + self.assertGreater(image.height, 200) + colors = image.convert("RGB").getcolors(maxcolors=1_000_000) + self.assertIsNotNone(colors) + self.assertGreater(len(colors or []), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/hash_testing/new_testing/test_probability_map.py b/hash_testing/new_testing/test_probability_map.py new file mode 100644 index 0000000..95c34a8 --- /dev/null +++ b/hash_testing/new_testing/test_probability_map.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +"""Behavior tests for probability_map.py.""" + +from __future__ import annotations + +import csv +import io +import os +import subprocess +import tempfile +import unittest +from contextlib import redirect_stderr +from pathlib import Path + +import generate_input +import probability_map + + +class PrepareHashTests(unittest.TestCase): + def test_compiles_c_source_when_binary_is_missing(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + hash_directory = root / "constant" + hash_directory.mkdir() + (hash_directory / "bin_hash.c").write_text( + "#include \n" + "int main(void) { puts(\"00000001\"); return 0; }\n", + encoding="utf-8", + ) + + executable = probability_map.prepare_hash("constant", root) + + self.assertEqual(executable, hash_directory / "bin_hash") + self.assertEqual( + subprocess.check_output([executable, "word"], text=True).strip(), + "00000001", + ) + + def test_adds_python_shebang_and_execute_permission(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + hash_directory = root / "python_hash" + hash_directory.mkdir() + source = hash_directory / "bin_hash.py" + source.write_text("print('00000002')\n", encoding="utf-8") + + executable = probability_map.prepare_hash("python_hash", root) + + self.assertEqual(executable, source) + self.assertTrue(os.access(source, os.X_OK)) + self.assertTrue( + source.read_text(encoding="utf-8").startswith( + "#!/usr/bin/env python3\n" + ) + ) + self.assertEqual( + subprocess.check_output([executable, "word"], text=True).strip(), + "00000002", + ) + + +class ProbabilityTests(unittest.TestCase): + def test_counts_changed_hash_bits_relative_to_source(self) -> None: + probabilities = probability_map.bit_probabilities( + source_hash=0b0000, + changed_hashes=[0b0001, 0b0011, 0b0010, 0b0000], + bits=4, + ) + + self.assertEqual(probabilities, [0.5, 0.5, 0.0, 0.0]) + + def test_csv_table_has_operation_rows_and_bit_columns(self) -> None: + stream = io.StringIO() + + probability_map.write_csv_table( + stream, + { + "replace": (4, [0.25, 0.75]), + "delete": (2, [0.5, 0.0]), + }, + ) + + rows = list(csv.reader(io.StringIO(stream.getvalue()))) + self.assertEqual(rows[0], ["operation", "pairs", "bit_0", "bit_1"]) + self.assertEqual(rows[1], ["replace", "4", "0.250000", "0.750000"]) + self.assertEqual(rows[2], ["delete", "2", "0.500000", "0.000000"]) + + def test_parser_accepts_multiple_source_words(self) -> None: + arguments = probability_map.build_parser().parse_args(["first", "second"]) + + self.assertEqual(arguments.words, ["first", "second"]) + + def test_aggregates_samples_from_multiple_words_and_warns_on_shortfall(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + executable = Path(temporary_directory) / "hash.py" + executable.write_text( + "#!/usr/bin/env python3\n" + "import sys\n" + "print(f'{sum(sys.argv[1].encode()):08x}')\n", + encoding="utf-8", + ) + executable.chmod(0o755) + warnings = io.StringIO() + + with redirect_stderr(warnings): + table = probability_map.build_probability_table( + executable=executable, + sources=["A", "B"], + operations=["delete"], + count=10, + operation_count=1, + alphabet=generate_input.DEFAULT_ALPHABET, + seed=1, + max_attempts=None, + ) + + pair_count, probabilities = table["delete"] + self.assertEqual(pair_count, 2) + self.assertEqual( + probabilities[:8], + [0.5, 0.5, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0], + ) + self.assertEqual(warnings.getvalue().count("delete"), 2) + + +class GenerateWordsTests(unittest.TestCase): + def test_count_is_an_upper_bound_when_unique_results_are_exhausted(self) -> None: + words = generate_input.generate_words( + source="abc", + count=100, + operation="delete", + operation_count=1, + seed=42, + ) + + self.assertEqual(set(words), {"ab", "ac", "bc"}) + + +if __name__ == "__main__": + unittest.main()