From 97a6c3d1cb21e88439115d01c3329b53a7b977f5 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Nov 2026 20:31:54 +0000 Subject: [PATCH 1/2] Fix leaks, round-trip failures and config bugs found by the second bug hunt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Masking: - Masks of military IDs, BR numbers, passports, tax numbers and units are regenerated until they differ from the original and from the open text (a military ID was masked as itself about once in a hundred). - Components of open numbers (1.2, 2024-2025) count as open text, so a numeric mask cannot match part of them. - Names in brackets, slashes or asterisks are recognised; the closing bracket is no longer swallowed by the patronymic mask. - Infinitives (Видати, Надати) are not taken for surnames. - Ranks after * or / are found with all their words. - Custom-pattern and always_mask masks are held as placeholders until the name pass is over, so they are not masked twice; protected phrases are skipped by custom patterns. - Text dates keep the month's case and do not start inside a longer number. Unmasking: - JSON strings are unmasked as one text, matching how masking numbers mask occurrences across the document; masking also treats them as one text. - Text-date months are restored in the case found in the masked text. Configuration: - exclusions match all apostrophe variants; the local template no longer erases shared lists; max_name_length is null in the shared config. - ENV booleans with unknown words are skipped with a warning; ENV preserve_case wins over the file; router_rules filters combine with enable_*; password_generation strings, fractional integers, re-mask warn counts and empty sections are handled. - hash_algorithm case, backup_suffix separators, non-string logging.level and non-string section keys give errors instead of tracebacks. - diagnose returns 1 for unreadable or malformed mapping files. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XT6iUWaQgahXDB9TWX9Bq7 Generated-With: Claude Code 2.1.42 --- CHANGELOG.md | 79 ++++++++ config.yaml | 6 +- config_example.py | 2 +- config_example.yaml | 6 +- config_local.example.yaml | 17 +- datamasking/_jsontext.py | 57 ++++++ datamasking/diagnose.py | 43 +++-- datamasking/extras/config.py | 60 ++++-- datamasking/extras/masking_logger.py | 2 +- datamasking/masking/cli.py | 63 +++++-- datamasking/masking/constants.py | 11 +- datamasking/masking/context.py | 25 +-- datamasking/masking/engine.py | 42 ++++- datamasking/masking/exclusions.py | 41 +++- datamasking/masking/helpers.py | 29 ++- datamasking/masking/language.py | 12 +- datamasking/masking/mask_military.py | 82 ++++---- datamasking/masking/mask_personal.py | 37 ++-- datamasking/unmasking/engine.py | 28 ++- tests/test_config_findings.py | 272 +++++++++++++++++++++++++++ tests/test_fuzz_findings2.py | 209 ++++++++++++++++++++ 21 files changed, 987 insertions(+), 136 deletions(-) create mode 100644 datamasking/_jsontext.py create mode 100644 tests/test_config_findings.py create mode 100644 tests/test_fuzz_findings2.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 095b614..116aeea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). (`Рапорт Білого Ярослава Миколайовича`). - `Шановна Любове Андріївно!`: the vocative `Любове` was not recognised, so the name got a surname mask and the patronymic a male name. +- A name in brackets, slashes or asterisks (`(Петренко Іван Петрович)`, + `/Петренко І.П./`) kept the surname open, masked the name as a surname and + dropped the closing bracket from the masked text. The rank in + `(сержант Петренко)` was not masked either. +- An infinitive opening an order item was taken for a surname when no + patronymic followed: `Видати Петренку Івану` became + `Вильчи Павлу Хомович`, with the surname masked as a first name. + +### Fixed — numbers left unmasked +- A mask could equal the original, and the number stayed open. Masks of + military IDs change two digits out of six, so about one ID in a hundred + (`АА 585285`) was not masked. A one-digit BR number (`№5дск`) was not + masked one time in ten. Passport numbers, tax numbers (ІПН) and military + units were also affected, more rarely. Each of these masks is now + regenerated until it differs from the original and from the open text. + Masks that already differed stay the same as before. ### Fixed — unmasking did not restore the original - A rank `… медичної служби` masked as `генерал-майор медичної служби` was @@ -49,6 +65,69 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). are now recorded in text order. - A brigade number broken across lines lost the rest of its name in the masked text (`60 окремої\nмеханізованої бригади` → `30 окремої`). +- JSON files: unmasking counted the occurrences of each mask anew in every + string, while masking numbers them across the whole document. When two + different values in different strings got the same mask (`№ 2` and `№ 6` + both masked as `№ 4`), the second one came back as the first. Masking also + checked a new mask only against the open words of its own string. Both now + treat all the strings of a file as one text. +- An order number mask could equal part of an open number (`№ 2` masked as + `1` next to `п. 1.2`), and unmasking gave its value to the wrong place. +- A naval rank after `*` or `/` (`*старший матрос Кравченко …*`) was masked + without its first word, and the result (`*старший моряк`) was not restored. +- A text date with a capitalised month came back in lower case + (`06 Жовтня 2025` → `06 жовтня 2025`). A month in capitals is now masked in + capitals and restored in capitals too. The end of a longer number was taken + for a day (`123 травня 2025` masked as `125 серпня 2026`) and was not + restored. +- A value listed in `exclusions.always_mask` or matched by a custom pattern + with `action: mask` was masked once more as a surname (`сержант Сокіл` → + `Цуліж` → `Цулій`), and unmasking returned the first mask instead of + `Сокіл`. A custom pattern such as `[A-Z]{3,}` also matched the placeholder + of a protected phrase, which then disappeared from both texts. + +### Fixed — configuration +- `exclusions` (words, phrases, always_mask, remove) now match every + apostrophe: `Дем'ян` in the list also covers `Дем’ян` and `Демʼян`. Before, + `always_mask` left those spellings open, and `phrases`/`words` did not + protect them. +- `config_local.example.yaml` set every `exclusions` list to `[]`. A copy of + the template therefore erased the lists from the shared `config.yaml`, + `always_mask` included. The template now uses `null`, which keeps the + shared value. +- `config.yaml` and `config_example.yaml` set `validation.max_name_length: 50`, + so with them a longer double surname stayed open, while without a + configuration it was masked. The value is now `null` (no limit), and masks + with the shared file are again the same as without it. +- An environment variable with an unknown boolean word turned the option off: + `DATA_MASKING_ENCRYPT_OUTPUT=y` wrote a plaintext mapping even with + `encrypt_output: true` in the file. Such a value is now skipped with a + warning, like an invalid number. +- `DATA_MASKING_PRESERVE_CASE` lost to `masking_rules.preserve_case` from the + file; the environment now wins, as for every other key. +- `router_rules.skip_types` / `only_types` in a file made the same file's + `masking_rules.enable_*` switches ignored. Both now apply: a type is masked + only when neither turns it off. `--only` / `--exclude` on the command line + still override `enable_*`. +- `security.password_generation: "false"` (a quoted string) was ignored and + replaced the whole section. It is now read as a boolean, and any other + string is an error. +- A fractional value of an integer key (`hash_digest_size: 2.9`) was cut down + silently; it is now an error. +- With `--re-mask N`, the count of matches left open by a custom pattern with + `action: warn` was multiplied by N. +- An empty section (`exclusions:` with only comments, `masking_rules: {}`) + was reported as a key with no effect, and `strict_mode` stopped the run. +- Configuration errors that ended in a traceback now give a message and exit + code 1: `system.hash_algorithm` in capitals (`SHA256`), `backup_suffix` + with a path separator, `logging.level` that is not a string, and a section + key that is not a string (YAML `on:`, `yes:`, a number). An unknown + `logging.level` (`verbose`) now gives a warning instead of silently + becoming `INFO`. +- `python -m datamasking.diagnose` returned 0 for an unreadable or encrypted + file and for a JSON array, and stopped with a traceback on + `{"mappings": null}` or on an entry without `masked_as`. It now reports the + problem and returns 1. ### Changed — masks - Naval ranks get naval masks: `капітан 1-го рангу` and `капітан-лейтенант` diff --git a/config.yaml b/config.yaml index aac7bd4..bc0d2ff 100644 --- a/config.yaml +++ b/config.yaml @@ -244,8 +244,10 @@ validation: # Дозволити скорочене по батькові (І.І.) allow_abbreviated_patronymic: true - # Обмеження довжини імен - max_name_length: 50 + # Обмеження довжини імен (max_name_length: null — без обмеження, як без + # конфігурації; до 3.1.21 тут стояло 50, і довше подвійне прізвище з + # цим файлом лишалось відкритим) + max_name_length: null min_name_length: 3 # -------------------------------------------------------------------------- diff --git a/config_example.py b/config_example.py index 475c5dd..489ee70 100644 --- a/config_example.py +++ b/config_example.py @@ -107,7 +107,7 @@ class ValidationConfig: validate_rank_dictionary: bool = True # лише True strict_pib_format: bool = False allow_abbreviated_patronymic: bool = True - max_name_length: int = 50 + max_name_length: Optional[int] = None # None — без обмеження min_name_length: int = 3 diff --git a/config_example.yaml b/config_example.yaml index 9841218..b6d7624 100644 --- a/config_example.yaml +++ b/config_example.yaml @@ -240,8 +240,10 @@ validation: # Дозволити скорочене по батькові (І.І.) allow_abbreviated_patronymic: true - # Обмеження довжини імен - max_name_length: 50 + # Обмеження довжини імен (max_name_length: null — без обмеження, як без + # конфігурації; до 3.1.21 тут стояло 50, і довше подвійне прізвище з + # цим файлом лишалось відкритим) + max_name_length: null min_name_length: 3 # -------------------------------------------------------------------------- diff --git a/config_local.example.yaml b/config_local.example.yaml index 29a774a..fe7532b 100644 --- a/config_local.example.yaml +++ b/config_local.example.yaml @@ -16,14 +16,17 @@ # ========================================================================== # Приклад: власні позивні маскуються завжди, місцеві абревіатури й назви — -# ніколи. Списки exclusions ДОДАЮТЬСЯ до переліків dictionaries з config.yaml. +# ніколи. Списки exclusions ДОДАЮТЬСЯ до переліків dictionaries з config.yaml, +# але ЗАМІНЮЮТЬ однойменні списки exclusions з config.yaml (повторіть тут +# спільні записи). null — лишити список із config.yaml (до 3.1.21 тут стояло +# [], і скопійований шаблон стирав спільні always_mask, phrases …). exclusions: - abbreviations: [] # напр. [ТРО, ОК] - words: [] # напр. [Рада, Раді] - phrases: [] # напр. ["Верховн* Рад*"] - legal_acts: [] # напр. ["розпорядженн* президента"] - always_mask: [] # напр. ["Грім", "Сокіл"] - remove: [] # прибрати запис із dictionaries + abbreviations: null # напр. [ТРО, ОК] + words: null # напр. [Рада, Раді] + phrases: null # напр. ["Верховн* Рад*"] + legal_acts: null # напр. ["розпорядженн* президента"] + always_mask: null # напр. ["Грім", "Сокіл"] + remove: null # прибрати запис із dictionaries # Приклад: інші значення спільних ключів (розкоментуйте потрібне) # system: diff --git a/datamasking/_jsontext.py b/datamasking/_jsontext.py new file mode 100644 index 0000000..58237cd --- /dev/null +++ b/datamasking/_jsontext.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Рядкові значення JSON як один текст (v3.1.21). + +Номери входжень масок (instance tracking) при маскуванні JSON спільні для +всього документа, а unmask до 3.1.21 обробляв кожен рядок окремо, з +лічильниками від нуля: якщо два різні оригінали в різних рядках діставали +одну маску («№ 2» і «№ 6» → «№ 4»), другий відновлювався першим. Перевірка +колізій маски з відкритим текстом теж бачила лише свій рядок. Тому і mask, +і unmask обробляють усі рядки разом — через роздільник, якого немає в +жодному з них (окремий рядок із символу приватної зони Unicode: не літера, +не цифра, не пробіл, тож жоден шаблон його не перетинає). + +Ключі словників не маскуються й сюди не входять. Лише stdlib — модуль +імпортують і mask, і unmask. +""" +from typing import Any, Iterator, List, Optional + + +def collect_strings(data: Any) -> List[str]: + """Рядкові значення (не ключі) у порядку обходу dict/list.""" + out: List[str] = [] + + def walk(node: Any) -> None: + if isinstance(node, dict): + for value in node.values(): + walk(value) + elif isinstance(node, list): + for item in node: + walk(item) + elif isinstance(node, str): + out.append(node) + + walk(data) + return out + + +def replace_strings(data: Any, values: Iterator[str]) -> Any: + """Копія *data*, де рядкові значення по черзі замінено на *values*.""" + if isinstance(data, dict): + return {key: replace_strings(value, values) for key, value in data.items()} + if isinstance(data, list): + return [replace_strings(item, values) for item in data] + if isinstance(data, str): + return next(values) + return data + + +def separator(strings: List[str]) -> Optional[str]: + """Роздільник, якого немає в жодному рядку (None — якщо всі символи + приватної зони вже зайняті).""" + for code in range(0xE000, 0xF900): + ch = chr(code) + if not any(ch in s for s in strings): + return "\n" + ch + "\n" + return None diff --git a/datamasking/diagnose.py b/datamasking/diagnose.py index e538bc5..0457a8e 100644 --- a/datamasking/diagnose.py +++ b/datamasking/diagnose.py @@ -91,10 +91,24 @@ def load_json(path: Path) -> Optional[Dict]: try: with open(path, 'r', encoding='utf-8') as f: data = json.load(f) - return data if isinstance(data, dict) else None except (FileNotFoundError, PermissionError, OSError, json.JSONDecodeError, UnicodeDecodeError) as e: print(f"❌ Помилка читання JSON {path}: {e}") + if Path(path).suffix.lower() == '.enc': + print(" Це зашифрований mapping: diagnose читає лише .json") return None + # Поле mappings — словник категорій, кожна категорія — словник (до 3.1.21 + # {"mappings": null} чи масив замість об'єкта давали traceback або exit 0) + mappings = data.get("mappings", {}) if isinstance(data, dict) else None + if not isinstance(mappings, dict) or not all(isinstance(v, dict) for v in mappings.values()): + print(f"❌ {path}: це не файл маппінгу (очікується об'єкт із полем mappings)") + return None + result: Dict = data + return result + + +def _masked_as(value: Any) -> Any: + """Маска запису v1 (рядок) або v2 (словник; без masked_as — None).""" + return value.get("masked_as") if isinstance(value, dict) else value def analyze_category_structure(category_name: str, items: Dict[str, Any]) -> Dict[str, Any]: @@ -142,7 +156,7 @@ def analyze_category_structure(category_name: str, items: Dict[str, Any]) -> Dic return {"count": len(items), "format": "unknown", "has_instances": False} -def compare_mappings(file1: Path, file2: Path) -> None: +def compare_mappings(file1: Path, file2: Path) -> bool: """ Порівнює два JSON файли маппінгу (Drift Check). @@ -157,8 +171,8 @@ def compare_mappings(file1: Path, file2: Path) -> None: data1 = load_json(file1) data2 = load_json(file2) - if not data1 or not data2: - return + if data1 is None or data2 is None: + return False # Отримання метаданих t1 = datetime.fromtimestamp(file1.stat().st_mtime).strftime('%Y-%m-%d %H:%M:%S') @@ -198,8 +212,8 @@ def compare_mappings(file1: Path, file2: Path) -> None: common_keys = set(items1.keys()) & set(items2.keys()) for key in common_keys: # Нормалізація для порівняння v1 та v2 - val1 = items1[key]["masked_as"] if isinstance(items1[key], dict) else items1[key] - val2 = items2[key]["masked_as"] if isinstance(items2[key], dict) else items2[key] + val1 = _masked_as(items1[key]) + val2 = _masked_as(items2[key]) if val1 != val2: drift_count += 1 @@ -214,9 +228,10 @@ def compare_mappings(file1: Path, file2: Path) -> None: print(" Це означає, що детермінованість порушена (або змінено seed/hash/алгоритм).") else: print("✅ Маски стабільні (для спільних ключів значення однакові).") + return True -def diagnose_single_file(map_path: str) -> None: +def diagnose_single_file(map_path: str) -> bool: """ Виводить детальну інформацію про один файл маппінгу без порівняння. Використовується, коли в історії знайдено лише один файл. @@ -224,7 +239,7 @@ def diagnose_single_file(map_path: str) -> None: map_file = Path(map_path) if not map_file.exists(): print(f"❌ ПОМИЛКА: Файл не знайдено: {map_file}") - return + return False print("=" * 80) print("📋 ДІАГНОСТИКА MASKING MAP") @@ -232,7 +247,8 @@ def diagnose_single_file(map_path: str) -> None: print(f"\n📂 Файл: {map_file}") data = load_json(map_file) - if not data: return + if data is None: + return False print(f"🔢 Версія: {data.get('version', 'N/A')}") print(f"🕐 Timestamp: {data.get('timestamp', 'N/A')}") @@ -251,6 +267,7 @@ def diagnose_single_file(map_path: str) -> None: if analysis.get("has_instances"): stats = analysis.get("instance_stats", {}) print(f" ✓ Instance tracking: {stats.get('total_instances', 0)} посилань") + return True # ============================================================================ @@ -449,7 +466,7 @@ def main() -> int: # АВТОМАТИЧНИЙ РЕЖИМ: Порівняти останній (0) з передостаннім (1) if len(latest_maps) < 2: if latest_maps: - diagnose_single_file(str(latest_maps[0])) + return 0 if diagnose_single_file(str(latest_maps[0])) else 1 else: print("❌ Файлів masking_map_*.json не знайдено.") return 1 @@ -474,8 +491,7 @@ def main() -> int: if not Path(arg).is_file(): print(f"❌ Файл не знайдено: {arg}") return 1 - diagnose_single_file(arg) - return 0 + return 0 if diagnose_single_file(arg) else 1 elif len(files_input) >= 2: # Явне порівняння двох вказаних файлів @@ -488,7 +504,8 @@ def main() -> int: if not target_file_a.exists() or not target_file_b.exists(): print("❌ Один з файлів маппінгу не існує.") return 1 - compare_mappings(target_file_a, target_file_b) + if not compare_mappings(target_file_a, target_file_b): + return 1 return 0 diff --git a/datamasking/extras/config.py b/datamasking/extras/config.py index 8bac2e7..d376067 100644 --- a/datamasking/extras/config.py +++ b/datamasking/extras/config.py @@ -388,6 +388,11 @@ def leaves(value: Any, prefix: str) -> List[str]: out: List[str] = [] for section, values in data.items(): + # Порожня відома секція («exclusions:» із самими коментарями, + # «masking_rules: {}») — нічого не задає, але й не помилка (до + # 3.1.21 strict_mode через неї завершувався з кодом 1) + if str(section) in EFFECTIVE_KEYS and values in (None, {}): + continue out += leaves(values, str(section)) return out @@ -441,6 +446,13 @@ def format_ignored_keys_warning(source: str, keys: List[str], limit: int = 10) - f"config_example.yaml or regenerate it: data-mask --init-config") +def _has_field(obj: Any, key: Any) -> bool: + """Ключ — поле dataclass-секції. До 3.1.21 тут був hasattr(): ключ-число + чи булеве (YAML «on:», «yes:») давав TypeError з traceback, а «__class__» + чи назва методу вважались полем.""" + return isinstance(key, str) and key in getattr(obj, "__dataclass_fields__", {}) + + _TRUE_WORDS = frozenset({"true", "yes", "on", "1"}) _FALSE_WORDS = frozenset({"false", "no", "off", "0"}) @@ -532,12 +544,12 @@ def from_dict(cls, data: Dict[str, Any]) -> "Config": if "system" in data and isinstance(data["system"], dict): for k, v in data["system"].items(): - if hasattr(cfg.system, k): + if _has_field(cfg.system, k): _set(cfg.system, k, v, "system") if "password_generation" in data and isinstance(data["password_generation"], dict): for k, v in data["password_generation"].items(): - if hasattr(cfg.password_generation, k): + if _has_field(cfg.password_generation, k): _set(cfg.password_generation, k, v, "password_generation") if "security" in data and isinstance(data["security"], dict): @@ -545,48 +557,50 @@ def from_dict(cls, data: Dict[str, Any]) -> "Config": for k, v in sec.items(): if k == "password_generation" and isinstance(v, dict): for pk, pv in v.items(): - if hasattr(cfg.security.password_generation, pk): + if _has_field(cfg.security.password_generation, pk): _set(cfg.security.password_generation, pk, pv, "security.password_generation") - elif k == "password_generation" and isinstance(v, bool): + elif k == "password_generation" and v is not None: # Шаблон 2.x писав `password_generation: true` — bool - # замість секції; не підміняємо dataclass булевим - cfg.security.password_generation.enabled = v - elif hasattr(cfg.security, k): + # замість секції; не підміняємо dataclass булевим (і + # рядком: до 3.1.21 "false" у лапках замінював секцію, + # і пароль усе одно генерувався) + _set(cfg.security.password_generation, "enabled", v, "security.password_generation") + elif _has_field(cfg.security, k): _set(cfg.security, k, v, "security") if "masking_rules" in data and isinstance(data["masking_rules"], dict): for k, v in data["masking_rules"].items(): - if hasattr(cfg.masking_rules, k): + if _has_field(cfg.masking_rules, k): _set(cfg.masking_rules, k, v, "masking_rules") if "validation" in data and isinstance(data["validation"], dict): for k, v in data["validation"].items(): - if hasattr(cfg.validation, k): + if _has_field(cfg.validation, k): _set(cfg.validation, k, v, "validation") if "router_rules" in data and isinstance(data["router_rules"], dict): for k, v in data["router_rules"].items(): - if hasattr(cfg.router_rules, k): + if _has_field(cfg.router_rules, k): _set(cfg.router_rules, k, v, "router_rules") if "logging" in data and isinstance(data["logging"], dict): for k, v in data["logging"].items(): - if hasattr(cfg.logging, k): + if _has_field(cfg.logging, k): _set(cfg.logging, k, v, "logging") if "remask" in data and isinstance(data["remask"], dict): for k, v in data["remask"].items(): - if hasattr(cfg.remask, k): + if _has_field(cfg.remask, k): _set(cfg.remask, k, v, "remask") if "exclusions" in data and isinstance(data["exclusions"], dict): for k, v in data["exclusions"].items(): - if hasattr(cfg.exclusions, k): + if _has_field(cfg.exclusions, k): _set(cfg.exclusions, k, v, "exclusions") if "dictionaries" in data and isinstance(data["dictionaries"], dict): for k, v in data["dictionaries"].items(): - if hasattr(cfg.dictionaries, k): + if _has_field(cfg.dictionaries, k): _set(cfg.dictionaries, k, v, "dictionaries") return cfg @@ -800,7 +814,19 @@ def _apply_env(self) -> None: value: Any if typ is bool: - value = raw.lower() in ("1", "true", "yes", "on") + # Як у файлі (_set_coerced): невідоме слово не є false. До + # 3.1.21 DATA_MASKING_ENCRYPT_OUTPUT=y вимикало шифрування, + # навіть коли у файлі стояло true + word = raw.strip().lower() + if word in _TRUE_WORDS: + value = True + elif word in _FALSE_WORDS: + value = False + else: + logger.warning( + "Invalid boolean for %s=%r (use true/false) — skipped", env_var, raw + ) + continue elif typ is int: try: value = int(raw) @@ -816,6 +842,10 @@ def _apply_env(self) -> None: if section_obj is not None and hasattr(section_obj, attr): setattr(section_obj, attr, value) logger.debug("ENV override: %s.%s = %r", section, attr, value) + if env_var == "DATA_MASKING_PRESERVE_CASE": + # masking_rules.preserve_case з файлу перекриває + # system.preserve_case — але не ENV (до 3.1.21 перекривав) + self._config.masking_rules.preserve_case = None # ----- CLI argument overlay ----- diff --git a/datamasking/extras/masking_logger.py b/datamasking/extras/masking_logger.py index b3a53f5..75f91b4 100644 --- a/datamasking/extras/masking_logger.py +++ b/datamasking/extras/masking_logger.py @@ -131,7 +131,7 @@ def __init__( log_file: Optional path to a log file. """ self.logger = logging.getLogger(name) - self.logger.setLevel(getattr(logging, level.upper(), logging.INFO)) + self.logger.setLevel(getattr(logging, str(level).strip().upper(), logging.INFO)) self.format_type = format_type # Internal statistics counters diff --git a/datamasking/masking/cli.py b/datamasking/masking/cli.py index 44df996..b37b868 100644 --- a/datamasking/masking/cli.py +++ b/datamasking/masking/cli.py @@ -360,6 +360,14 @@ def _setup_logger(args, config) -> Optional[Any]: if log_level is None: log_level = "INFO" + # logging.level / DATA_MASKING_LOG_LEVEL: до 3.1.21 число чи true давали + # traceback, а невідоме слово («verbose») мовчки ставало INFO + levels = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") + if str(log_level).strip().upper() not in levels: + _config_warning(args, f"logging.level must be one of {', '.join(levels)}, " + f"got {log_level!r} — using INFO") + log_level = "INFO" + log_level = str(log_level).strip().upper() console = getattr(log_cfg, 'log_to_console', True) is not False # logging.format (рядок у файлі логу), max_log_size_mb + log_rotation_count @@ -498,7 +506,8 @@ def _or_default(value, default): def _int_setting(value, name: str, lo: int, hi: int) -> int: """Ціле в межах [lo, hi]; інакше ValueError з текстом для користувача.""" - if isinstance(value, bool): + # 2.9 — не ціле (до 3.1.21 мовчки ставало 2) + if isinstance(value, bool) or (isinstance(value, float) and not value.is_integer()): raise ValueError(f"{name} must be an integer, got {value!r}") try: n = int(value) @@ -580,14 +589,19 @@ def _apply_level1_settings_impl(args, config, logger) -> Optional[str]: # Текстові дати, прізвища, по батькові, номери документів: окремо або # «як відповідний загальний прапорець» - if not getattr(args, 'only', None) and not getattr(args, 'exclude', None): - def follow(key: str, base: bool) -> bool: + # (router_rules.only_types / skip_types — разом з ними, як enable_* вище) + selective = getattr(args, 'only', None) or getattr(args, 'exclude', None) + narrow = getattr(args, '_selective_source', None) is not None + if not selective or narrow: + def follow(key: str, base: bool, current: bool) -> bool: value = getattr(rules, key, None) - return base if value is None else bool(value) - _cfg.MASK_DATE_TEXT = follow('enable_date_text', _cfg.MASK_DATES) - _cfg.MASK_SURNAMES = follow('enable_surnames', _cfg.MASK_NAMES) - _cfg.MASK_PATRONYMICS = follow('enable_patronymics', _cfg.MASK_NAMES) - _cfg.MASK_DOCUMENT_NUMBERS = follow('enable_document_numbers', _cfg.MASK_ORDERS) + result = base if value is None else bool(value) + return result and current if narrow else result + _cfg.MASK_DATE_TEXT = follow('enable_date_text', _cfg.MASK_DATES, _cfg.MASK_DATE_TEXT) + _cfg.MASK_SURNAMES = follow('enable_surnames', _cfg.MASK_NAMES, _cfg.MASK_SURNAMES) + _cfg.MASK_PATRONYMICS = follow('enable_patronymics', _cfg.MASK_NAMES, _cfg.MASK_PATRONYMICS) + _cfg.MASK_DOCUMENT_NUMBERS = follow('enable_document_numbers', _cfg.MASK_ORDERS, + _cfg.MASK_DOCUMENT_NUMBERS) # Перемаскування if remask is not None: @@ -702,7 +716,13 @@ def _apply_config_settings(args, config, logger) -> Optional[str]: only_types = getattr(args, 'only', None) if SELECTIVE_AVAILABLE else None exclude_types = getattr(args, 'exclude', None) if SELECTIVE_AVAILABLE else None - if not only_types and not exclude_types: + # --only / --exclude з командного рядка важать більше за enable_*. + # router_rules.only_types / skip_types — того самого рівня, що й + # enable_*: діють обидва (тип маскується, лише якщо його не вимкнено + # ні там, ні там). До 3.1.21 skip_types у файлі вимикав дію всіх + # enable_* того самого файлу + from_config = getattr(args, '_selective_source', None) is not None + if (not only_types and not exclude_types) or from_config: config_rules_map = { 'enable_ranks': 'MASK_RANKS', 'enable_names': 'MASK_NAMES', @@ -718,7 +738,7 @@ def _apply_config_settings(args, config, logger) -> Optional[str]: for config_key, global_var in config_rules_map.items(): value = getattr(masking_rules, config_key, None) if value is not None: - setattr(_cfg, global_var, bool(value)) + setattr(_cfg, global_var, bool(value) and (getattr(_cfg, global_var) or not from_config)) if logger: logger.debug(f"Config: {config_key} = {value}") @@ -733,12 +753,22 @@ def _apply_config_settings(args, config, logger) -> Optional[str]: if cfg_preserve_case is not None: _cfg.PRESERVE_CASE = bool(cfg_preserve_case) + # Суфікс копії — частина імені файлу: «/x» до 3.1.21 давав traceback + # (Path.with_name) лише в момент запису, коли вихідний файл уже існував + cfg_suffix = getattr(system_cfg, 'backup_suffix', None) + if cfg_suffix is not None and (not isinstance(cfg_suffix, str) + or any(c in cfg_suffix for c in '/\\\0')): + return (f"system.backup_suffix must be a file name suffix without path separators, " + f"got {cfg_suffix!r}") + cfg_hash = getattr(system_cfg, 'hash_algorithm', None) - if cfg_hash is not None and str(cfg_hash).lower() not in ('blake2b', 'md5', 'sha1', 'sha256', 'sha512'): + if cfg_hash is not None and str(cfg_hash).strip().lower() not in ('blake2b', 'md5', 'sha1', 'sha256', 'sha512'): return (f"system.hash_algorithm must be one of blake2b, md5, sha1, sha256, sha512, " f"got {cfg_hash!r}") if cfg_hash is not None: - _cfg.HASH_ALGORITHM = str(cfg_hash) + # Як перевірено — без урахування регістру: до 3.1.21 «SHA256» + # проходило перевірку, а get_deterministic_seed падав з traceback + _cfg.HASH_ALGORITHM = str(cfg_hash).strip().lower() # system.faker_locale — словники faker для синтетичних прізвищ/імен locale = getattr(system_cfg, 'faker_locale', None) @@ -948,6 +978,10 @@ def _run_multi_pass_masking(input_data, is_json: bool, masking_dict: Dict, chain = MappingChain() masked_data = input_data + # Збіги шаблонів з action: warn — за документом, а не × проходи (до + # 3.1.21 кожен прохід знаходив ті самі збіги ще раз) + from datamasking.masking import custom as _custom + warn_counts: Dict[str, int] = {} for pass_num in range(1, re_mask_passes + 1): print(f" Прохід {pass_num}/{re_mask_passes}...") @@ -966,6 +1000,11 @@ def _run_multi_pass_masking(input_data, is_json: bool, masking_dict: Dict, for category, mappings in pass_dict["mappings"].items(): pass_dict["statistics"][category] = len(mappings) chain.add_pass(pass_dict) + if pass_num == 1: + warn_counts = dict(_custom.WARN_COUNTS) + + _custom.WARN_COUNTS.clear() + _custom.WARN_COUNTS.update(warn_counts) # Звіт — за першим проходом (справжні унікальні значення документа); # до 3.1.10 статистика підсумовувалась по всіх проходах (×N) і diff --git a/datamasking/masking/constants.py b/datamasking/masking/constants.py index 5759814..e0bdb09 100644 --- a/datamasking/masking/constants.py +++ b/datamasking/masking/constants.py @@ -188,10 +188,10 @@ def set_faker_locale(locale: str) -> None: ) DATE_TEXT_PATTERN = re.compile( r'(?:["\u201c\u201e«]?\s*)?' # optional opening quote - r'(\d{1,2})' # day + r'(? None: # Лапки всіх стилів (українські «», німецькі „“, англійські "" '', ‟). # Значення в лапках («сержант», «Петренко») мають розпізнаватись і маскуватись. QUOTE_CHARS = '«»„“”‟“”„"\'' +# Дужки та інші обрамлення слова: «(Петренко Іван Петрович)», «/Коваль/». +# До 3.1.21 «(Петренко» не вважалось словом ПІБ (прізвище лишалось +# відкритим), а «Петрович)» вважалось — разом із дужкою, що зникала з тексту +BRACKET_CHARS = '()[]{}<>' +WRAP_CHARS = QUOTE_CHARS + BRACKET_CHARS + '*/' MAX_INPUT_FILE_SIZE = 100 * 1024 * 1024 -EXCLUDE_WORDS_LOWER = frozenset(w.lower() for w in EXCLUDE_WORDS) +EXCLUDE_WORDS_LOWER = _exclusions.apostrophe_variants(w.lower() for w in EXCLUDE_WORDS) RANKS_LIST_LOWER = frozenset(r.lower() for r in RANKS_LIST) # Усі граматичні форми звань (для точного збігу вмісту лапок «...») ALL_RANK_FORMS_LOWER = frozenset(f.lower() for f in ALL_RANK_FORMS) diff --git a/datamasking/masking/context.py b/datamasking/masking/context.py index 4f7e039..de000ac 100644 --- a/datamasking/masking/context.py +++ b/datamasking/masking/context.py @@ -12,7 +12,7 @@ from datamasking.masking import constants as _cfg from datamasking.masking.declension import VOCATIVE, analyze_name, analyze_patronymic -from datamasking.masking.helpers import normalize_string, normalize_identifier, is_pib_anchor +from datamasking.masking.helpers import normalize_string, normalize_identifier, is_pib_anchor, strip_word, ends_clause from datamasking.masking.language import looks_like_name, detect_gender_by_patronymic, is_likely_surname_by_case # Роздільники слів усередині рядка: будь-які пробільні символи (пробіл, таб, @@ -31,7 +31,7 @@ def _strip_word(word: str) -> str: - return word.strip(_cfg.QUOTE_CHARS).strip(',.!?;:') + return strip_word(word, leading_punct=True) def is_patronymic_word(word: str, prev_word: Optional[str] = None) -> bool: @@ -315,7 +315,10 @@ def parse_hybrid_line(line: str) -> Tuple[Optional[str], Optional[str], Optional char_tok: List[int] = [] # символ normalized_line → індекс у parts (-1 — роздільник) tok_start: Set[int] = set() # позиції, з яких починається слово parts for i, part in enumerate(parts): - norm = normalize_string(part) + # «*старший матрос …*», «/старшому моряку …/»: без цього звання + # знаходилось без першого слова («матрос» → «моряк»), а unmask бачив + # у «*старший моряк» інше, двослівне звання й не відновлював його + norm = normalize_string(part.strip('*/')) if not norm: continue if chunks: @@ -367,7 +370,7 @@ def parse_hybrid_line(line: str) -> Tuple[Optional[str], Optional[str], Optional rank_matches.sort(key=lambda x: x[0]) rank_index, found_rank, rank_position, rank_word_count = rank_matches[0] pib_start_index = rank_position + rank_word_count - rank_words = [w.strip(_cfg.QUOTE_CHARS) for w in + rank_words = [strip_word(w) for w in parts[rank_position:rank_position + rank_word_count]] found_rank_original_case = ' '.join(rank_words) if rank_words else found_rank @@ -469,14 +472,14 @@ def _caps_block_without_patronymic(parts: List[str], start: int) -> bool: if not _caps_word(parts[start]): return False begin = start - while begin > 0 and _caps_word(parts[begin - 1]) and parts[begin - 1].rstrip(_cfg.QUOTE_CHARS)[-1:] not in ',.;:!?': + while begin > 0 and _caps_word(parts[begin - 1]) and not ends_clause(parts[begin - 1]): begin -= 1 run: List[str] = [] for word in parts[begin:]: if not _caps_word(word): break run.append(_strip_word(word)) - if word.rstrip(_cfg.QUOTE_CHARS)[-1:] in ',.;:!?': + if ends_clause(word): break if len(run) < 3: return False @@ -548,13 +551,13 @@ def _adjacent_rank(parts: List[str], start: int, rank_matches) -> str: регістрі оригіналу), або "" — якщо його там немає. rank_matches містить лише перше входження кожної форми, тож спершу дивимось на самі слова.""" for k in range(min(4, start), 0, -1): - words = [w.strip(_cfg.QUOTE_CHARS).strip(',.;:!?') for w in parts[start - k:start]] + words = [strip_word(w, leading_punct=True) for w in parts[start - k:start]] phrase = ' '.join(words) if phrase.lower() in _cfg.RANKS_LIST_LOWER: return phrase for _idx, rank_form, pos, count in rank_matches: if pos + count == start: - words = [w.strip(_cfg.QUOTE_CHARS) for w in parts[pos:pos + count]] + words = [strip_word(w) for w in parts[pos:pos + count]] return ' '.join(words) if words else rank_form return "" @@ -563,7 +566,7 @@ def _rank_word_as_surname(word: str, next_word: Optional[str]) -> bool: """«капітан Майор Іван Іванович»: слово-звання з великої літери одразу після справжнього звання і перед іменем — це прізвище (Майор, Сотник, Полковник — реальні українські прізвища).""" - clean = word.strip(_cfg.QUOTE_CHARS).rstrip(',.!?;:') + clean = strip_word(word) if len(clean) < 3 or clean.lower() not in _cfg.RANKS_LIST_LOWER: return False if not (clean[0].isupper() and clean[1:].islower()): @@ -576,7 +579,7 @@ def _rank_word_as_surname(word: str, next_word: Optional[str]) -> bool: def _extract_pib_words(parts, start: int, after_rank: bool): pib_words = [] for idx, word in enumerate(parts[start:start + 3]): - clean = word.strip(_cfg.QUOTE_CHARS).rstrip(',.!?;:') + clean = strip_word(word) if looks_like_name(word): pib_words.append(clean) elif idx == 0 and after_rank and _rank_word_as_surname( @@ -587,6 +590,6 @@ def _extract_pib_words(parts, start: int, after_rank: bool): # Розділовий знак після слова завершує ПІБ: «Сергійович, ІПН …» — # інакше наступне слово з великої приклеювалось, і зібраний рядок # не існував у тексті (заміна не спрацьовувала) - if word.rstrip(_cfg.QUOTE_CHARS)[-1:] in ',.;:!?': + if ends_clause(word): break return pib_words diff --git a/datamasking/masking/engine.py b/datamasking/masking/engine.py index addd614..5cca764 100644 --- a/datamasking/masking/engine.py +++ b/datamasking/masking/engine.py @@ -8,12 +8,14 @@ """ import bisect +import copy import functools import json import random import re from typing import Any, Callable, Dict, List, Optional, Pattern, Tuple, Union +from datamasking import _jsontext from datamasking.masking import constants as _cfg from datamasking.masking import surname as _surname from datamasking.masking import custom as _custom @@ -337,7 +339,6 @@ def _unmasked_residue(text: str, masking_dict: Dict, instance_counters: Dict, Пробний прогін на копіях mapping і лічильників: скільки разів слово трапляється в тексті, мінус скільки разів його замасковано.""" - import copy rank_counts: Dict[str, int] = {} if _cfg.MASK_RANKS: for m in _get_rank_forms_re().finditer(text): @@ -370,6 +371,12 @@ def _unmasked_residue(text: str, masking_dict: Dict, instance_counters: Dict, rank_residue = {base for base, n in rank_counts.items() if n > masked.get(base, 0)} token_residue = {tok for tok, n in token_counts.items() if n > masked.get(tok, 0)} + # Числова маска шукається при unmask з межею «не цифра», тож «1» знаходиться + # і всередині відкритого «п. 1.2» / «2024-2025». До 3.1.21 тут був лише + # токен цілком, маска наказу «№ 1» могла збігтися з частиною такого числа, + # і unmask віддавав її номер не тому входженню + for tok in [t for t in token_residue if t[:1].isdigit()]: + token_residue.update(re.findall(r'\d+', tok)) return rank_residue, token_residue @@ -894,6 +901,14 @@ def _add_skip(item) -> None: items_to_skip.append(item) covered_skip[item['start']:item['end']] = b'\x01' * (item['end'] - item['start']) + # Захищені фрази (___KEEP…___) не маскуються жодною фазою: до 3.1.21 + # custom-шаблон «[A-Z]{3,}» знаходив «KEEPA» в токені, і фраза зникала + for token, _original in kept or (): + pos = text.find(token) + while pos != -1: + covered_skip[pos:pos + len(token)] = b'\x01' * len(token) + pos = text.find(token, pos + len(token)) + if not _cfg.MASK_DATES: for match in re.finditer(_cfg.UKRAINIAN_DATE_PATTERN, text): _add_skip({'start': match.start(), 'end': match.end(), 'text': match.group(0), 'reason': 'full_date', 'type': 'date'}) @@ -1017,6 +1032,10 @@ def _phase_date_text() -> None: segments = [] prev_end = 0 + # Маски custom-шаблонів і always_mask — за токенами до кінця розбору + # ПІБ: до 3.1.21 «сержант Сокіл» → «сержант Цуліж», далі фаза ПІБ + # маскувала маску ще раз («Цулій»), і unmask не повертав «Сокіл» + custom_held: List[Tuple[str, str]] = [] for item in items_to_mask: if item['start'] < prev_end: continue # перекриття — пропускаємо if text[item['start']:item['end']] != item['full_text']: continue @@ -1034,6 +1053,10 @@ def _phase_date_text() -> None: replacement = _mask_date_text(item['full_text'], masking_dict, instance_counters) elif item['type'] == 'custom': replacement = _custom.mask_custom(item['full_text'], masking_dict, instance_counters, text) + if replacement and replacement != item['full_text']: + token = "___CMASK" + _letters(len(custom_held)) + "___" + custom_held.append((token, replacement)) + replacement = token elif item['type'] == 'order_simple': masked = mask_order_number(item['number_part'], masking_dict, instance_counters) replacement = item['full_text'].replace(item['number_part'], masked, 1) @@ -1192,6 +1215,8 @@ def _replace_span(line: str, span: Tuple[int, int, List[str]], token: str) -> st # основного циклу, з пропуском уже замаскованих форм text = _mask_quoted_ranks(text, masking_dict, instance_counters) + for token, value in custom_held: + text = text.replace(token, value, 1) if line_join: text = text.replace(_LINE_JOIN, '\n') return text @@ -1201,6 +1226,21 @@ def mask_json_recursive(data: Any, masking_dict: Dict, instance_counters: Dict) # не збіглась зі словом з іншого поля _surname.enter_document(json.dumps(data, ensure_ascii=False) if isinstance(data, (dict, list)) else str(data)) try: + # Усі рядки — одним текстом (див. datamasking/_jsontext.py): так + # unmask рахує входження масок так само, а маска не збігається з + # відкритим словом іншого рядка + strings = _jsontext.collect_strings(data) + sep = _jsontext.separator(strings) if len(strings) > 1 else None + if sep is not None: + saved = copy.deepcopy(masking_dict), dict(instance_counters) + parts = mask_text_context_aware(sep.join(strings), masking_dict, instance_counters).split(sep) + if len(parts) == len(strings): + return _jsontext.replace_strings(data, iter(parts)) + # Маскування зачепило роздільник (не мало б статися) — по рядку + masking_dict.clear() + masking_dict.update(saved[0]) + instance_counters.clear() + instance_counters.update(saved[1]) return _mask_json_recursive_impl(data, masking_dict, instance_counters) finally: _surname.exit_document() diff --git a/datamasking/masking/exclusions.py b/datamasking/masking/exclusions.py index c77500e..a50e384 100644 --- a/datamasking/masking/exclusions.py +++ b/datamasking/masking/exclusions.py @@ -112,14 +112,36 @@ _WORD_EDGE_AFTER = r"(?![\w'’ʼ])" +_APOSTROPHES = "'’ʼ" +_APOSTROPHE_CLASS = "[" + _APOSTROPHES + "]" + + +def _escape_word(word: str) -> str: + # Апостроф — будь-який із трьох: у документах трапляються всі (до 3.1.21 + # «Дем'ян» у переліку не знаходив «Дем’ян» і «Демʼян») + return "".join(_APOSTROPHE_CLASS if ch in _APOSTROPHES else re.escape(ch) for ch in word) + + +def apostrophe_variants(words: Iterable[str]) -> FrozenSet[str]: + """Слова з усіма варіантами апострофа (' ’ ʼ): перевірки «слово in + перелік» знаходять будь-яке написання.""" + out = set() + for w in words: + out.add(w) + if any(a in w for a in _APOSTROPHES): + base = w.replace("’", "'").replace("ʼ", "'") + out.update(base.replace("'", a) for a in _APOSTROPHES) + return frozenset(out) + + def phrase_regex(phrase: str) -> str: """«Кабінет* Міністрів» → ``Кабінет\\w*\\s+Міністрів`` (без меж слова).""" parts = [] for word in phrase.split(): if word.endswith("*"): - parts.append(re.escape(word[:-1]) + r"\w*") + parts.append(_escape_word(word[:-1]) + r"\w*") else: - parts.append(re.escape(word)) + parts.append(_escape_word(word)) return r"\s+".join(parts) @@ -198,15 +220,17 @@ def build(section: Any = None, dictionaries: Any = None) -> Exclusions: """Переліки (dictionaries з config.yaml або вбудовані) + секція exclusions конфігурації. Raises ValueError.""" lists = {key: _items(section, key) for key in KEYS} - remove = {r.lower() for r in lists["remove"]} + remove = apostrophe_variants(r.lower() for r in lists["remove"]) base_abbreviations = _base(dictionaries, "abbreviations", BUILTIN_ABBREVIATIONS) base_words = _base(dictionaries, "non_name_words", BUILTIN_WORDS) base_acts = _base(dictionaries, "legal_acts", BUILTIN_LEGAL_ACTS) - abbreviations = frozenset(a.lower() for a in base_abbreviations if a.lower() not in remove) \ - | frozenset(a.lower() for a in lists["abbreviations"]) - words = frozenset(w.lower() for w in base_words if w.lower() not in remove) \ - | frozenset(w.lower() for w in lists["words"]) + abbreviations = apostrophe_variants( + [a.lower() for a in base_abbreviations if a.lower() not in remove] + + [a.lower() for a in lists["abbreviations"]]) + words = apostrophe_variants( + [w.lower() for w in base_words if w.lower() not in remove] + + [w.lower() for w in lists["words"]]) acts = tuple(a for a in base_acts if a.lower() not in remove) \ + tuple(lists["legal_acts"]) if acts: @@ -233,7 +257,8 @@ def build(section: Any = None, dictionaries: Any = None) -> Exclusions: def describe(ex: Exclusions) -> str: """Текст для --list-exclusions (діючі переліки).""" def block(title: str, items: Iterable[str]) -> List[str]: - items = sorted(items) + # Варіанти апострофа (apostrophe_variants) — одним записом + items = sorted(i for i in items if "’" not in i and "ʼ" not in i) return [f"{title} ({len(items)}):", " " + (", ".join(items) if items else "—"), ""] lines: List[str] = [] lines += block("Abbreviations (never masked as a surname)", ex.abbreviations) diff --git a/datamasking/masking/helpers.py b/datamasking/masking/helpers.py index 3031b12..f0785bf 100644 --- a/datamasking/masking/helpers.py +++ b/datamasking/masking/helpers.py @@ -97,8 +97,9 @@ def normalize_string(s: str) -> str: if not s: return "" s_str = str(s).lower() s_str = s_str.replace("'", "'").replace('\xa0', ' ').replace(".", " ").replace(";", " ") - # Лапки → пробіл, щоб значення в лапках («сержант») знаходились під час пошуку - for q in _cfg.QUOTE_CHARS: + # Лапки й дужки → пробіл, щоб значення в лапках («сержант») і дужках + # («(сержант …)») знаходились під час пошуку + for q in _cfg.QUOTE_CHARS + _cfg.BRACKET_CHARS: s_str = s_str.replace(q, ' ') return re.sub(r'\s+', ' ', s_str).strip() @@ -107,9 +108,31 @@ def normalize_identifier(identifier: str) -> str: identifier_str = str(identifier).upper() return re.sub(r'[\s\-.]', '', identifier_str).strip() +_PUNCT = ',.!?;:' + + +def strip_word(word: str, leading_punct: bool = False) -> str: + """Слово без обрамлення (лапки, дужки, «/…/», «*…*») і розділових знаків + у кінці — у будь-якому поєднанні («Петровича),»); з *leading_punct* — + і розділових знаків на початку.""" + lead = _cfg.WRAP_CHARS + (_PUNCT if leading_punct else "") + trail = _cfg.WRAP_CHARS + _PUNCT + start, end = 0, len(word) + while start < end and word[start] in lead: + start += 1 + while end > start and word[end - 1] in trail: + end -= 1 + return word[start:end] + + +def ends_clause(word: str) -> bool: + """Після слова — розділовий знак або закривна дужка: ПІБ на ньому закінчується.""" + return word.rstrip(_cfg.QUOTE_CHARS + '*/')[-1:] in _PUNCT + ')]}>' + + def is_pib_anchor(word: str) -> bool: if word.startswith("___") and word.endswith("___"): return False - word = word.strip(_cfg.QUOTE_CHARS).strip(",.!?;:") + word = strip_word(word, leading_punct=True) if not word or len(word) <= 2: return False # Слова з цифрами/= — не ПІБ (ІПН=3698521592, «138», кодування) if re.search(r'[\d=]', word): return False diff --git a/datamasking/masking/language.py b/datamasking/masking/language.py index 56da301..c1e70e5 100644 --- a/datamasking/masking/language.py +++ b/datamasking/masking/language.py @@ -15,19 +15,20 @@ from datamasking.masking.declension import ( VOCATIVE, analyze_name, analyze_patronymic, decline_name, ) -from datamasking.masking.helpers import get_deterministic_seed +from datamasking.masking.helpers import get_deterministic_seed, strip_word def is_likely_surname_by_case(word: str) -> bool: if word.startswith("___") and word.endswith("___"): return False - word = word.strip(_cfg.QUOTE_CHARS) + word = strip_word(word) if not word or len(word) < 3: return False letters_only = re.sub(r"[-']", '', word) return letters_only.isupper() and len(letters_only) >= 3 def looks_like_name(word: str) -> bool: if word.startswith("___") and word.endswith("___"): return False - clean_word = word.strip(_cfg.QUOTE_CHARS).rstrip(',.!?;:') + clean_word = strip_word(word) + if not clean_word: return False # validation.min_name_length / max_name_length (за замовчуванням 3 / без меж) if len(clean_word) < _cfg.NAME_MIN_LENGTH: return False if _cfg.NAME_MAX_LENGTH and len(clean_word) > _cfg.NAME_MAX_LENGTH: return False @@ -36,6 +37,11 @@ def looks_like_name(word: str) -> bool: # ніколи не імена/прізвища if clean_word.lower().endswith(('ємо', 'имо', 'емо', 'єте', 'ите', 'ете')): return False + # Інфінітиви на початку пункту наказу (Видати, Надати, Зарахувати, + # Виключити) — теж: до 3.1.21 «Видати Петренку Івану» без по батькові + # розбиралось як «прізвище Видати, ім'я Петренку, по батькові Івану» + if len(clean_word) >= 5 and clean_word.lower().endswith(('ати', 'яти', 'ити', 'іти', 'їти', 'ути', 'тися', 'тись')): + return False # Віддієслівні та абстрактні іменники (-ння/-ття/-ість: Відділення, # Призначення, Відповідальність) — ні імена, ні прізвища (v3.1.17) if len(clean_word) >= 5 and clean_word.lower().endswith(('ння', 'ття', 'ість')): diff --git a/datamasking/masking/mask_military.py b/datamasking/masking/mask_military.py index b5092e2..8dafb84 100644 --- a/datamasking/masking/mask_military.py +++ b/datamasking/masking/mask_military.py @@ -31,22 +31,34 @@ def mask_military_unit(original: str, masking_dict: Dict, instance_counters: Dic match = re.match(r'^([А-ЯA-Z])( ?)(\d{4})$', original) if not match: return original letter, sep = match.group(1), match.group(2) - seed = get_deterministic_seed(original) - random.seed(seed) - digits = ''.join([str(random.randint(0, 9)) for _ in range(4)]) - masked = letter + sep + digits + + def _gen(attempt: int) -> str: + _reseed(original, attempt) + return letter + sep + ''.join(str(random.randint(0, 9)) for _ in range(4)) + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "military_unit", original, masked) + +def _reseed(original: str, attempt: int) -> None: + """Seed генератора для спроби *attempt* (0 — як до появи повторних спроб, + тож перша маска не змінюється).""" + random.seed(get_deterministic_seed(original) if attempt == 0 else get_deterministic_seed(f"{original}\x00{attempt}")) + + def _redigit(original: str, attempt: int = 0) -> str: """Замінює всі цифри випадковими (детерміновано від оригіналу й номера спроби).""" - random.seed(get_deterministic_seed(original) if attempt == 0 else get_deterministic_seed(f"{original}\x00{attempt}")) + _reseed(original, attempt) return re.sub(r'\d', lambda _: str(random.randint(0, 9)), original) def _distinct_mask(original: str, generate: "Callable[[int], str]", attempts: int = 10) -> str: """Маска, що не дорівнює оригіналу і не збігається з числом/словом документа (інакше unmask мінял би їх місцями, v3.1.9). *generate(attempt)* - — детермінований генератор; після *attempts* спроб — остання маска.""" + — детермінований генератор; після *attempts* спроб — остання маска. + + Потрібна кожному генератору, що міняє лише частину цифр: до 3.1.21 + військовий квиток (змінюються 2 цифри з 6) маскувався сам у себе в ~1 % + випадків, «№5дск» — у кожному десятому, і номер лишався відкритим.""" masked = generate(0) for attempt in range(1, attempts + 1): if masked != original and not _surname.document_contains(masked): @@ -83,9 +95,6 @@ def mask_br_number(original: str, masking_dict: Dict, instance_counters: Dict) - if original in masking_dict["mappings"]["br_number"]: masked = masking_dict["mappings"]["br_number"][original]["masked_as"] else: - seed = get_deterministic_seed(original) - random.seed(seed) - # Витягуємо суфікс (дск, п, к) suffix_match = re.search(r'(дск|п|к)$', original) suffix = suffix_match.group(1) if suffix_match else "" @@ -102,27 +111,29 @@ def mask_br_number(original: str, masking_dict: Dict, instance_counters: Dict) - # Обробляємо частини розділені слешами parts = re.split(r'(/)', number_part) - masked_parts = [] - for part in parts: - if part == '/': masked_parts.append(part) - elif part and re.match(r'^\d', part): - digit_match = re.match(r'(\d+)(.*)', part) - if digit_match: - digits = digit_match.group(1) - rest = digit_match.group(2) - masked_digits = ''.join([str(random.randint(0, 9)) for _ in range(len(digits))]) - masked_parts.append(masked_digits + rest) + + def _gen(attempt: int) -> str: + _reseed(original, attempt) + masked_parts = [] + for part in parts: + if part == '/': masked_parts.append(part) + elif part and re.match(r'^\d', part): + digit_match = re.match(r'(\d+)(.*)', part) + if digit_match: + digits = digit_match.group(1) + rest = digit_match.group(2) + masked_digits = ''.join([str(random.randint(0, 9)) for _ in range(len(digits))]) + masked_parts.append(masked_digits + rest) + else: masked_parts.append(part) else: masked_parts.append(part) - else: masked_parts.append(part) - masked = prefix + ''.join(masked_parts) + suffix + return prefix + ''.join(masked_parts) + suffix + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "br_number", original, masked) def mask_br_number_slash(original: str, masking_dict: Dict, instance_counters: Dict) -> str: if original in masking_dict["mappings"]["br_number_slash"]: masked = masking_dict["mappings"]["br_number_slash"][original]["masked_as"] else: - seed = get_deterministic_seed(original) - random.seed(seed) suffix_match = re.search(r'(дск|п|к)$', original) suffix = suffix_match.group(1) if suffix_match else "" prefix = "" @@ -135,13 +146,17 @@ def mask_br_number_slash(original: str, masking_dict: Dict, instance_counters: D if suffix: number_part = number_part[:-len(suffix)] parts = number_part.split('/') - masked_parts = [] - for part in parts: - if part.strip().isdigit(): - masked_parts.append(''.join([str(random.randint(0, 9)) for _ in range(len(part.strip()))])) - else: - masked_parts.append(part) - masked = prefix + '/'.join(masked_parts) + suffix + + def _gen(attempt: int) -> str: + _reseed(original, attempt) + masked_parts = [] + for part in parts: + if part.strip().isdigit(): + masked_parts.append(''.join([str(random.randint(0, 9)) for _ in range(len(part.strip()))])) + else: + masked_parts.append(part) + return prefix + '/'.join(masked_parts) + suffix + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "br_number_slash", original, masked) def mask_br_number_complex(original: str, masking_dict: Dict, instance_counters: Dict) -> str: @@ -292,9 +307,10 @@ def mask_date_text(original: str, masking_dict: Dict, instance_counters: Dict) - parts = masked_key.split() if len(parts) < 3: return original - # Місяць — у регістрі оригіналу («Жовтня» → «Грудня»); unmask порівнює - # без урахування регістру й повертає оригінальний - new_month_text = parts[1].capitalize() if month_name[:1].isupper() else parts[1] + # Місяць — у регістрі оригіналу («Жовтня» → «Грудня», «ЖОВТНЯ» → + # «ГРУДНЯ»): mapping зберігає дату в нижньому регістрі, і unmask бере + # регістр місяця з маски (до 3.1.21 капс ставав «Грудня») + new_month_text = _apply_original_case(month_name, parts[1]) return original.replace(day, parts[0], 1).replace( month_name, new_month_text, 1).replace(year, parts[2], 1) diff --git a/datamasking/masking/mask_personal.py b/datamasking/masking/mask_personal.py index 47595bb..4d54d1e 100644 --- a/datamasking/masking/mask_personal.py +++ b/datamasking/masking/mask_personal.py @@ -24,6 +24,7 @@ ) from datamasking.masking import surname as _surname from datamasking.masking.surname import synthesize_surname, known_surname_forms +from datamasking.masking.mask_military import _distinct_mask, _reseed _IPN_WEIGHTS = (-1, 5, 7, 9, 4, 6, 10, 5, 7) @@ -49,15 +50,15 @@ def mask_ipn(original: str, masking_dict: Dict, instance_counters: Dict) -> str: masked = masking_dict["mappings"]["ipn"][original]["masked_as"] else: if len(original) != 10 or not original.isdigit(): return original - seed = get_deterministic_seed(original) - random.seed(seed) - middle = ''.join([str(random.randint(0, 9)) for _ in range(6)]) - masked = original[:3] + middle + original[-1] # validation.validate_ipn_checksum: валідний ІПН → валідна маска # (контрольна цифра перераховується); невалідний — як і раніше - if _cfg.VALIDATE_IPN_CHECKSUM and is_valid_ipn(original): - first9 = original[:3] + middle - masked = first9 + str(ipn_checksum(first9)) + fix_checksum = _cfg.VALIDATE_IPN_CHECKSUM and is_valid_ipn(original) + + def _gen(attempt: int) -> str: + _reseed(original, attempt) + first9 = original[:3] + ''.join(str(random.randint(0, 9)) for _ in range(6)) + return first9 + (str(ipn_checksum(first9)) if fix_checksum else original[-1]) + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "ipn", original, masked) def mask_passport_id(original: str, masking_dict: Dict, instance_counters: Dict) -> str: @@ -71,10 +72,11 @@ def mask_passport_id(original: str, masking_dict: Dict, instance_counters: Dict) masked = masking_dict["mappings"]["passport_id"][original]["masked_as"] else: if len(original) != 9 or not original.isdigit(): return original - seed = get_deterministic_seed(original) - random.seed(seed) - middle = ''.join([str(random.randint(0, 9)) for _ in range(5)]) - masked = original[:3] + middle + original[-1] + + def _gen(attempt: int) -> str: + _reseed(original, attempt) + return original[:3] + ''.join(str(random.randint(0, 9)) for _ in range(5)) + original[-1] + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "passport_id", original, masked) def mask_military_id(original: str, masking_dict: Dict, instance_counters: Dict) -> str: @@ -96,11 +98,14 @@ def mask_military_id(original: str, masking_dict: Dict, instance_counters: Dict) prefix = prefix_match.group(1) or "" sep = prefix_match.group(2) if prefix else "" digits = prefix_match.group(3) - seed = get_deterministic_seed(original) - random.seed(seed) - middle = ''.join([str(random.randint(0, 9)) for _ in range(2)]) - masked_digits = digits[:2] + middle + digits[-2:] - masked = prefix + sep + masked_digits + + # Змінюються лише 2 середні цифри — без повторних спроб маска + # збігалась з оригіналом у кожному сотому квитку (до 3.1.21) + def _gen(attempt: int) -> str: + _reseed(original, attempt) + middle = ''.join(str(random.randint(0, 9)) for _ in range(2)) + return prefix + sep + digits[:2] + middle + digits[-2:] + masked = _distinct_mask(original, _gen) return add_to_mapping(masking_dict, instance_counters, "military_id", original, masked) def mask_surname(original: str, masking_dict: Dict, instance_counters: Dict) -> str: diff --git a/datamasking/unmasking/engine.py b/datamasking/unmasking/engine.py index 35983e8..4103be7 100644 --- a/datamasking/unmasking/engine.py +++ b/datamasking/unmasking/engine.py @@ -26,6 +26,7 @@ class FoundRank(TypedDict, total=False): text: str simple: bool +from datamasking import _jsontext from datamasking.rank_data import ( RANK_DECLENSIONS, RANK_FEMININE_MAP, @@ -283,6 +284,7 @@ def _lookup(matched: str, key: str): info["masked_as"] for info in mappings_copy.get("date_text", {}).values() if isinstance(info, dict) and "masked_as" in info } + date_text_lower = {m.lower() for m in date_text_masks} def _alternative(mask: str) -> str: if mask in date_text_masks: @@ -329,7 +331,9 @@ def _bounded(mask: str) -> str: matched = m.group(0) key = matched.lower() inst, counter_key = _lookup(matched, key) - quoted_date = False + # Текстова дата: mapping зберігає її в нижньому регістрі, тож регістр + # місяця («Жовтня», «ЖОВТНЯ») — з тексту (до 3.1.21 губився) + quoted_date = bool(inst) and key in date_text_lower if not inst and date_text_masks: # «28» вересня 2025 → ключ «28 вересня 2025» key = _DT_STRIP_RE.sub('', matched).lower() @@ -448,16 +452,30 @@ def unmask_json_with_stats(masked_data: Any, masking_map: Dict, map_version: str JSON давав 0/0, і strict_mode завжди завершувався помилкою).""" stats = {"restored_count": 0, "skipped_count": 0} + def unmask_one(text: str) -> Tuple[str, Dict]: + if map_version.startswith("v2"): + return unmask_text_v2(text, masking_map, map_version) + return unmask_text_v1(text, masking_map) + + # Усі рядки — одним текстом, як при маскуванні (datamasking/_jsontext.py): + # номери входжень масок спільні для всього документа. До 3.1.21 кожен + # рядок розмасковувався з нуля, і однакова маска двох різних оригіналів + # у різних рядках відновлювалась першим із них + strings = _jsontext.collect_strings(masked_data) + sep = _jsontext.separator(strings) if len(strings) > 1 else None + if sep is not None: + restored, st = unmask_one(sep.join(strings)) + parts = restored.split(sep) + if len(parts) == len(strings): + return _jsontext.replace_strings(masked_data, iter(parts)), st + def walk(data: Any) -> Any: if isinstance(data, dict): return {k: walk(v) for k, v in data.items()} if isinstance(data, list): return [walk(item) for item in data] if isinstance(data, str): - if map_version.startswith("v2"): - restored, st = unmask_text_v2(data, masking_map, map_version) - else: - restored, st = unmask_text_v1(data, masking_map) + restored, st = unmask_one(data) stats["restored_count"] += st.get("restored_count", 0) stats["skipped_count"] += st.get("skipped_count", 0) return restored diff --git a/tests/test_config_findings.py b/tests/test_config_findings.py new file mode 100644 index 0000000..6259711 --- /dev/null +++ b/tests/test_config_findings.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Знахідки аудиту конфігурації (після 3.1.20). + + - маску custom-шаблону / always_mask фаза ПІБ маскувала ще раз, і unmask + не повертав оригінал; шаблон «[A-Z]{3,}» знаходив службовий токен + захищеної фрази, і фраза зникала; + - exclusions не знаходили слів з іншим апострофом (’ ʼ замість '); + - скопійований шаблон config_local стирав спільні exclusions; + - невідоме булеве значення в ENV ставало false (шифрування вимикалось); + - config.yaml (max_name_length: 50) маскував не так, як запуск без нього; + - traceback замість помилки конфігурації: hash_algorithm у верхньому + регістрі, backup_suffix зі «/», logging.level не рядок, ключ секції не + рядок; + - router_rules.skip_types вимикав enable_* того самого файлу; + masking_rules.preserve_case перекривав ENV; password_generation: + "false" ігнорувався; 2.9 у цілому ключі ставало 2; --re-mask множив + лічильник warn; порожня секція — «ключ без дії» (strict_mode падав); + - diagnose: битий файл давав exit 0 або traceback. +""" +import json +import os +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from datamasking import diagnose # noqa: E402 +from datamasking.extras.config import ConfigLoader, ignored_config_keys # noqa: E402 +from datamasking.masking import cli as mask_cli # noqa: E402 +from datamasking.masking import constants as _cfg # noqa: E402 +from datamasking.masking import exclusions as _excl # noqa: E402 +from datamasking.unmasking.engine import unmask_text_v2 # noqa: E402 +from datamasking.unmasking.helpers import check_mapping_version # noqa: E402 +from tests.test_initials import mask # noqa: E402 + +try: + import yaml + _HAS_YAML = True +except ImportError: + _HAS_YAML = False +needs_yaml = pytest.mark.skipif(not _HAS_YAML, reason="pyyaml not installed") + + +def _rt(text): + masked, mapping = mask(text) + restored, _ = unmask_text_v2(masked, mapping, check_mapping_version(mapping)) + assert restored == text, (masked, restored) + return masked, mapping + + +def _exclusions(section): + ex = _excl.build(section) + _cfg.ABBREVIATION_WHITELIST = ex.abbreviations + _cfg.EXCLUDE_WORDS_LOWER = ex.words_lower + _cfg.CUSTOM_PATTERNS = ex.skip_patterns + _cfg.CUSTOM_PATTERNS + ex.mask_patterns + + +@pytest.fixture +def run(tmp_path, monkeypatch, capsys): + monkeypatch.chdir(tmp_path) + for var in [v for v in os.environ if v.startswith("DATA_MASKING_")]: + monkeypatch.delenv(var, raising=False) + + def _run(config=None, extra=(), text="сержант Петренко Іван Іванович прибув.\n", out="out.txt"): + if config is not None: + (tmp_path / "config.py").write_text(f"CONFIG = {config!r}\n", encoding="utf-8") + (tmp_path / "in.txt").write_text(text, encoding="utf-8") + rc = mask_cli.main(["-i", "in.txt", "-o", out, "--no-report", *extra]) + cap = capsys.readouterr() + path = tmp_path / out + return rc, path.read_text(encoding="utf-8") if path.exists() else "", cap.out, cap.err + + return _run + + +class TestCustomMasksNotMaskedTwice: + @pytest.mark.parametrize("text", [ + "Доповідь подав сержант Сокіл.", + "Позивний Сокіл, сержант Сокіл, СОКІЛ.", + ]) + def test_always_mask(self, text): + _exclusions({"always_mask": ["Сокіл"]}) + masked, mapping = _rt(text) + assert "Сокіл" not in masked and "СОКІЛ" not in masked + assert set(mapping["mappings"].get("surname", {})) == set(), mapping["mappings"] + + def test_mask_pattern_does_not_hit_kept_phrase(self): + from datamasking.masking import custom as _custom + _cfg.CUSTOM_PATTERNS = _custom.compile_patterns(["[A-Z]{3,}"]) + _exclusions({"phrases": ["Верховн* Рад*"]}) + masked, mapping = _rt("Звернення до Верховна Рада, позивний ABC.") + assert masked.startswith("Звернення до Верховна Рада, позивний ") + assert "ABC" not in masked + assert set(mapping["mappings"]["custom"]) == {"ABC"} + + +class TestApostropheVariants: + TEXT = "Позивний Дем'ян, позивний Дем’ян, позивний Демʼян прибули." + + def test_always_mask(self): + _exclusions({"always_mask": ["Дем'ян"]}) + masked, _ = _rt(self.TEXT) + assert "Дем" not in masked + + @pytest.mark.parametrize("section", [ + {"phrases": ["Дем'янів Лаз"]}, + {"phrases": ["Дем’янів Лаз"]}, + {"words": ["Дем'янів"]}, + {"words": ["Демʼянів"]}, + ]) + def test_kept(self, section): + _exclusions(section) + text = "в урочище Дем'янів Лаз.\nв урочище Дем’янів Лаз.\nв урочище Демʼянів Лаз." + masked, _ = _rt(text) + assert masked == text + + def test_describe_lists_word_once(self): + assert "дем'янів" in _excl.describe(_excl.build({"words": ["Дем’янів"]})) + assert "дем’янів" not in _excl.describe(_excl.build({"words": ["Дем’янів"]})) + + def test_remove_any_spelling(self): + ex = _excl.build({"words": ["Дем'янів"], "remove": ["Дем’янів"]}) + assert "дем'янів" in ex.words_lower # remove діє лише на вбудовані + assert _excl.apostrophe_variants(["a'b"]) == {"a'b", "a’b", "aʼb"} + + +@needs_yaml +class TestLocalTemplate: + def test_keeps_shared_exclusions(self, tmp_path, monkeypatch, capsys): + monkeypatch.chdir(tmp_path) + data = yaml.safe_load((ROOT / "config.yaml").read_text(encoding="utf-8")) + data["exclusions"]["always_mask"] = ["Сокіл"] + (tmp_path / "config.yaml").write_text(yaml.safe_dump(data, allow_unicode=True), encoding="utf-8") + (tmp_path / "config_local.yaml").write_bytes((ROOT / "config_local.example.yaml").read_bytes()) + os.chmod(tmp_path / "config_local.yaml", 0o600) + (tmp_path / "in.txt").write_text("Доповідь подав сержант Сокіл.\n", encoding="utf-8") + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--no-report"]) == 0 + assert "Loaded local config from config_local.yaml" in capsys.readouterr().out + assert "Сокіл" not in (tmp_path / "out.txt").read_text(encoding="utf-8") + + def test_root_config_has_no_name_length_limit(self): + data = yaml.safe_load((ROOT / "config.yaml").read_text(encoding="utf-8")) + assert data["validation"]["max_name_length"] is None + + +class TestValues: + def test_env_bool_unknown_word_is_skipped(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "config.py").write_text( + "CONFIG = {'security': {'encrypt_output': True}}\n", encoding="utf-8") + monkeypatch.setenv("DATA_MASKING_ENCRYPT_OUTPUT", "y") + assert ConfigLoader().load().security.encrypt_output is True + monkeypatch.setenv("DATA_MASKING_ENCRYPT_OUTPUT", " No ") + assert ConfigLoader().load().security.encrypt_output is False + + def test_hash_algorithm_any_case(self, run): + rc, out, *_ = run({"system": {"hash_algorithm": "SHA256"}}) + assert rc == 0 and "Петренко" not in out + + def test_backup_suffix_with_separator(self, run): + run() + rc, _out, stdout, _err = run({"system": {"backup_enabled": True, "backup_suffix": "/x"}}, + extra=("--force",)) + assert rc == 1 and "system.backup_suffix" in stdout + + @pytest.mark.parametrize("level", [10, True, "verbose"]) + def test_logging_level(self, run, level): + rc, _out, _stdout, err = run({"logging": {"level": level}}) + assert rc == 0 and "logging.level must be one of" in err + + @pytest.mark.parametrize("section", [{1: True}, {True: 1}, {"__class__": "x"}]) + def test_non_field_keys(self, run, section): + rc, _out, _stdout, err = run({"masking_rules": section}) + assert rc == 0 and "have no effect" in err + + def test_non_field_key_unmask(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "config.py").write_text("CONFIG = {'logging': {1: 2}}\n", encoding="utf-8") + assert ConfigLoader().load() is not None + + @pytest.mark.parametrize("key, value", [ + ("system", {"hash_digest_size": 2.9}), + ("validation", {"min_name_length": 2.5}), + ]) + def test_fractional_int(self, run, key, value): + rc, _out, stdout, _err = run({key: value}) + assert rc == 1 and "must be an integer" in stdout + + +class TestSemantics: + TEXT = "Сержант Петренко Іван Іванович, ІПН 1234567890, дата 01.10.2025.\n" + + def test_skip_types_and_enable_flags(self, run): + rc, out, *_ = run({"masking_rules": {"enable_ipn": False, "enable_names": False}, + "router_rules": {"skip_types": ["date"]}}, text=self.TEXT) + assert rc == 0 + assert "Петренко Іван Іванович" in out and "1234567890" in out and "01.10.2025" in out + + def test_skip_types_still_excludes_enabled_type(self, run): + rc, out, *_ = run({"masking_rules": {"enable_dates": True, "enable_date_text": True}, + "router_rules": {"skip_types": ["date"]}}, text=self.TEXT) + assert rc == 0 and "01.10.2025" in out and "Петренко" not in out + + def test_cli_only_wins_over_enable_flags(self, run): + rc, out, *_ = run({"masking_rules": {"enable_ipn": False}}, text=self.TEXT, + extra=("--only", "ipn")) + assert rc == 0 and "1234567890" not in out and "Петренко" in out + + def test_env_preserve_case_beats_file(self, run, monkeypatch): + monkeypatch.setenv("DATA_MASKING_PRESERVE_CASE", "false") + rc, out, *_ = run({"masking_rules": {"preserve_case": True}}, + text="СЕРЖАНТ Петренко Іван Іванович прибув.\n") + assert rc == 0 and not out.startswith(("СЕРЖАНТ", "СТАРШИЙ", "МОЛОДШИЙ")) + + def test_password_generation_string(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "config.py").write_text( + "CONFIG = {'security': {'password_generation': 'false'}}\n", encoding="utf-8") + cfg = ConfigLoader().load() + assert cfg.security.password_generation.enabled is False + assert cfg.security.password_generation.min_digits is not None + + def test_password_generation_bad_string(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "config.py").write_text( + "CONFIG = {'security': {'password_generation': 'maybe'}}\n", encoding="utf-8") + with pytest.raises(Exception, match="password_generation.enabled"): + ConfigLoader().load() + + def test_remask_warn_counted_once(self, run): + rc, _out, _stdout, err = run( + {"masking_rules": {"custom_patterns": [{"pattern": "ТЕЛ-\\d{4}", "name": "phone", "action": "warn"}]}}, + text="Зв'язок: ТЕЛ-1234, ТЕЛ-5678.\n", extra=("--re-mask", "3")) + assert rc == 0 and "'phone': 2 match(es)" in err + + @pytest.mark.parametrize("data", [ + {"exclusions": None, "masking_rules": {}}, + {"system": {"strict_mode": True}, "exclusions": None}, + ]) + def test_empty_section(self, data): + assert ignored_config_keys(data) == [] + + +class TestDiagnose: + @pytest.mark.parametrize("content", [ + "{broken", "[1, 2]", '{"mappings": null}', '{"mappings": {"surname": [1]}}', "garbage", + ]) + def test_bad_file_exit_1(self, tmp_path, monkeypatch, content): + monkeypatch.chdir(tmp_path) + (tmp_path / "m.json").write_text(content, encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["diagnose", "m.json"]) + assert diagnose.main() == 1 + + def test_compare_without_masked_as(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + entry = {"mappings": {"surname": {"A": {"instances": [1]}}}} + for name in ("a.json", "b.json"): + (tmp_path / name).write_text(json.dumps(entry), encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["diagnose", "a.json", "b.json"]) + assert diagnose.main() == 0 + + def test_compare_bad_file(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "a.json").write_text('{"mappings": {}}', encoding="utf-8") + (tmp_path / "b.json").write_text("{broken", encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["diagnose", "a.json", "b.json"]) + assert diagnose.main() == 1 diff --git a/tests/test_fuzz_findings2.py b/tests/test_fuzz_findings2.py new file mode 100644 index 0000000..37ed93d --- /dev/null +++ b/tests/test_fuzz_findings2.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Знахідки другого прогону фазера (після 3.1.20): числові типи, JSON, +обрамлення слів. + + - маска військового квитка (змінюються 2 цифри з 6), БР-номера, паспорта, + ІПН, військової частини могла збігтися з оригіналом — номер лишався + відкритим (квиток — у ~1 % випадків, «№5дск» — у кожному десятому); + - маска номера наказу «1» збігалась із частиною відкритого «п. 1.2»; + - JSON: номери входжень масок спільні для документа, а unmask рахував + їх для кожного рядка з нуля; + - інфінітив на початку пункту («Видати Петренку Івану») ставав прізвищем; + - ПІБ у дужках: «(Петренко» не було словом ПІБ (прізвище лишалось + відкритим), а «Петрович)» було — разом із дужкою, що зникала; + - «*старший матрос …*»: звання знаходилось без першого слова, і unmask + не відновлював «*старший моряк»; + - текстова дата: місяць з великої чи капсом повертався малими літерами, + а «123 травня 2025» маскувалось як «23 травня» і не відновлювалось. +""" +import json +import random +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from datamasking.masking.engine import mask_json_recursive # noqa: E402 +from datamasking.masking.mask_military import ( # noqa: E402 + mask_br_number, mask_military_unit, +) +from datamasking.masking.mask_personal import ( # noqa: E402 + mask_ipn, mask_military_id, mask_passport_id, +) +from datamasking.unmasking.engine import unmask_json_with_stats, unmask_text_v2 # noqa: E402 +from datamasking.unmasking.helpers import check_mapping_version # noqa: E402 +from tests.test_initials import make_masking_dict, mask # noqa: E402 + + +def _rt(text): + masked, mapping = mask(text) + restored, _ = unmask_text_v2(masked, mapping, check_mapping_version(mapping)) + assert restored == text, (masked, restored) + return masked, mapping + + +def _absent(masked, *words): + for w in words: + assert w not in masked, (w, masked) + + +def _json_rt(data): + mapping = make_masking_dict() + counters = {} + masked = mask_json_recursive(data, mapping, counters) + mapping["instance_tracking"] = counters + # Як у CLI: замаскований JSON записується у файл і читається назад + masked = json.loads(json.dumps(masked, ensure_ascii=False)) + restored, stats = unmask_json_with_stats(masked, mapping, check_mapping_version(mapping)) + assert restored == data, (masked, restored) + return masked, stats + + +class TestMaskNeverEqualsOriginal: + @pytest.mark.parametrize("func, make", [ + (mask_military_id, lambda r: f"АА {r.randint(100000, 999999)}"), + (mask_military_id, lambda r: f"мт-{r.randint(100000, 999999)}"), + (mask_military_id, lambda r: str(r.randint(100000, 999999))), + (mask_passport_id, lambda r: str(r.randint(10 ** 8, 10 ** 9 - 1))), + (mask_ipn, lambda r: str(r.randint(10 ** 9, 10 ** 10 - 1))), + (mask_military_unit, lambda r: f"А{r.randint(1000, 9999)}"), + (mask_br_number, lambda r: f"№{r.randint(1, 99)}дск"), + (mask_br_number, lambda r: f"№БР-{r.randint(1, 9)}/{r.randint(1, 9)}-К"), + ]) + def test_random_values(self, func, make): + rng = random.Random(7) + for _ in range(1500): + original = make(rng) + mapping = make_masking_dict() + mapping["mappings"].setdefault("br_number", {}) + assert func(original, mapping, {}) != original, original + + @pytest.mark.parametrize("text, secret", [ + ("військовий квиток АА 585285", "585285"), + ("посвідчення мт-767843 військовий квиток АА 585285", "585285"), + ("військовий квиток КВ 740021", "740021"), + ]) + def test_military_id_in_text(self, text, secret): + masked, _ = _rt(text) + _absent(masked, secret) + + def test_first_mask_unchanged(self): + # Повторні спроби — лише коли перша маска не годиться: маски, що + # й так відрізнялись від оригіналу, лишились тими самими + # (значення — як у 3.1.20) + assert mask_military_id("АА 123456", make_masking_dict(), {}) == "АА 129256" + assert mask_military_unit("А1234", make_masking_dict(), {}) == "А9377" + assert mask_passport_id("123456789", make_masking_dict(), {}) == "123775719" + assert mask_ipn("1234567890", make_masking_dict(), {}) == "1233548730" + mapping = make_masking_dict() + mapping["mappings"].setdefault("br_number", {}) + assert mask_br_number("№БР-123/45-К", mapping, {}) == "№БР-632/58-К" + + +class TestNumberInsideOpenNumber: + @pytest.mark.parametrize("text", [ + "582 одиниць і п. 1.2 і військової частини В8363\n" + "військової частини К2059 і наказ № 2 та наказ № 2\n", + "п. 1.2 та 2024-2025 роки; наказ № 2 від сьогодні, наказ № 3, наказ № 4, " + "наказ № 5, наказ № 6", + ]) + def test_roundtrip(self, text): + _rt(text) + + +class TestJson: + def test_shared_mask_across_strings(self): + _json_rt({"orders": [f"наказ № {i}" for i in range(1, 25)]}) + + def test_open_number_in_another_string(self): + _json_rt({"b": "п. 4", "a": "наказ № 2", "c": ["наказ № 6", {"x": "п. 1.2"}], "n": 5, "z": None}) + + def test_names_across_strings(self): + masked, stats = _json_rt({ + "a": "Звернення до Іване Петровичу", + "b": "Надати відпустку солдату Петренку Івану Петровичу", + "c": "Шановний Іване Петровичу!", + }) + assert "Петренк" not in json.dumps(masked, ensure_ascii=False) + assert stats["skipped_count"] == 0 + + @pytest.mark.parametrize("data", [ + {"one": "солдат Петренко Іван Петрович"}, + ["", "солдат Петренко Іван Петрович", "", "Петренко"], + {"k": [], "m": {}, "s": "без даних"}, + "солдат Петренко Іван Петрович", + ]) + def test_shapes(self, data): + _json_rt(data) + + +class TestInfinitiveNotSurname: + @pytest.mark.parametrize("text, verb, secrets", [ + ("Видати Петренку Івану", "Видати", ("Петренк", "Івану")), + ("Надати Петренку Івану відпустку", "Надати", ("Петренк", "Івану")), + ("Наказати Коваленку Петру", "Наказати", ("Коваленк", "Петру")), + ("Зарахувати Мельника Олега до списків", "Зарахувати", ("Мельник", "Олега")), + ("Видати ПЕТРЕНКУ ІВАНУ ПЕТРОВИЧУ посвідчення", "Видати", ("ПЕТРЕНК", "ІВАНУ")), + ]) + def test_verb_kept(self, text, verb, secrets): + masked, _ = _rt(text) + assert masked.startswith(verb + " "), masked + _absent(masked, *secrets) + + +class TestBrackets: + @pytest.mark.parametrize("text, wrap", [ + ("(Петренко Іван Петрович)", "()"), + ("[Петренко Іван Петрович]", "[]"), + ("{Петренко Іван Петрович}", "{}"), + ("/Петренко Іван Петрович/", "//"), + ("*Петренко Іван Петрович*", "**"), + ("(сержант Петренко Іван Петрович)", "()"), + ("Наказ (щодо Петренка Івана Петровича) виконати", None), + ("сержант (Петренко Іван Петрович), ІПН 1234567890", None), + ("(сержант Петренко)", "()"), + ("(сержант Петренко Іван Петрович), (солдат Коваль Олег Олегович)", None), + ]) + def test_masked_and_kept(self, text, wrap): + masked, _ = _rt(text) + _absent(masked, "Петренк", "Іван", "Петрович", "Коваль") + if wrap: + assert masked[0] == wrap[0] and masked[-1] == wrap[1], masked + assert masked.count("(") == text.count("(") and masked.count(")") == text.count(")") + + def test_pib_ends_at_closing_bracket(self): + masked, mapping = _rt("(Петренко Іван Петрович) Коваль") + assert all(")" not in k for cat in mapping["mappings"].values() for k in cat) + + +class TestWrappedNavalRank: + @pytest.mark.parametrize("text", [ + "/старшому моряку Ярославу Юрійовичу Петренку оголосити подяку./", + "*старший матрос Кравченко Олександр Богданович доповів*", + "*старший матрос Бондар Ігор Андрійович, РНОКПП 8984085240*", + "року *старший матрос Пилип'юк Любов Богданівна,", + ]) + def test_roundtrip(self, text): + _rt(text) + + +class TestTextDates: + @pytest.mark.parametrize("text", [ + "06 Жовтня 2025 р.", + "06 ЖОВТНЯ 2025 РОКУ", + "«06» ЖОВТНЯ 2025 та «06» Жовтня 2025 і 06 жовтня 2025", + "«06» жовтня 2025 року та 06 жовтня 2025 року", + ]) + def test_month_case_restored(self, text): + masked, _ = _rt(text) + _absent(masked.lower(), "жовтня") + + @pytest.mark.parametrize("text", ["123 травня 2025", "1 травня 20251"]) + def test_not_part_of_longer_number(self, text): + masked, _ = _rt(text) + assert masked == text From 5a88ed0b53675d1d5eaa2369a89da8917058eafb Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Nov 2026 22:03:04 +0000 Subject: [PATCH 2/2] Fix command-line and file handling bugs found by the CLI audit - system.encoding: auto honours a BOM (UTF-8/16/32) and keeps it byte for byte; text that decodes only with NUL characters is an error, not cp1251 (a UTF-16 file was left unmasked with exit 0). - --only/--exclude also drive the numbers of other documents, like enable_orders. - The mapping records input_format; unmask uses it instead of the masked file's extension (older mappings still use the extension). JSON input with a UTF-8 BOM is accepted. - unmask applies the same size limit as mask. - --password with security.encrypt_output: true is accepted; a password that is not valid UTF-8 is rejected before any file is written. - -o pointing to a hard link of the input is refused. - Mappings with an empty mask are rejected. - The report shows the executed number of --re-mask passes. - Damaged format-2 .enc headers and --init-config into a missing folder give errors instead of tracebacks. - unmask returns 2 for wrong usage (unset --password-env variable, missing --map, --to-version out of range or on a non-chain mapping). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XT6iUWaQgahXDB9TWX9Bq7 Generated-With: Claude Code 2.1.42 --- CHANGELOG.md | 40 +++++++ datamasking/_textio.py | 24 +++- datamasking/extras/security.py | 14 ++- datamasking/extras/selective.py | 5 + datamasking/masking/cli.py | 57 +++++++-- datamasking/unmasking/cli.py | 46 ++++--- datamasking/unmasking/io.py | 8 +- tests/test_cli_findings.py | 205 ++++++++++++++++++++++++++++++++ tests/test_to_version.py | 4 +- 9 files changed, 368 insertions(+), 35 deletions(-) create mode 100644 tests/test_cli_findings.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 116aeea..63bd0e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -129,6 +129,46 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `{"mappings": null}` or on an entry without `masked_as`. It now reports the problem and returns 1. +### Fixed — command line and files +- With `system.encoding: auto`, a UTF-16 file was read as cp1251, so nothing + was masked and the run still exited with 0. A byte order mark (UTF-8, + UTF-16, UTF-32) now decides the encoding, and the output and the restored + file keep it byte for byte. Text that decodes only with NUL characters + (UTF-16 without a BOM) is now an error instead of being guessed as cp1251. +- `--only` / `--exclude` did not control the numbers of other documents + (`довідку № 4567`): `--only surname` still masked them, and + `--exclude order_number` did not stop it. They now follow `order_number`, + as with `enable_orders` in the configuration. +- Unmasking chose between JSON and text by the masked file's extension, while + masking chose by the input's extension. Masked text saved as `out.json` + could not be unmasked, and JSON saved as `.txt` came back only in part. The + mapping now records the input format (`input_format`); older mappings + still use the extension. +- A JSON input with a UTF-8 BOM was rejected; it is now read like a text + file with a BOM. +- `unmask` could not read files over 100 MB even when + `validation.max_input_size_mb` allowed them, while `mask` did read them. + Both now apply the same limit. +- `--password` (or `--password-env`) with `security.encrypt_output: true` in + the configuration was rejected as if `--encrypt` were missing. +- A password that is not valid UTF-8 (from a terminal in another encoding) + failed only when the mapping was being encrypted. By then the masked file + was already written, and its mapping was not. Such a password is now + rejected before any file is written. +- `-o` pointing to a hard link of the input, with `--force`, overwrote the + original document; it is now refused like `-o` equal to the input. +- A mapping entry with an empty mask made unmasking insert the original + value between all characters of the text; such a mapping is now rejected. +- With `--re-mask 50`, the report said 50 passes, although the run was capped + at `remask.max_passes` (10). It now shows the passes actually made. +- Tracebacks replaced with error messages (exit code 1): a damaged header of a + format-2 `.enc` file, and `--init-config -c` with a missing folder or a + folder as the target. +- `unmask` now exits with 2 (wrong usage), as `mask` does, when + `--password-env` names an unset variable, when no `--map` is given for a + file that is not named `output_*`, and when `--to-version` is out of range + or used with a mapping that is not a chain. Before, it exited with 1. + ### Changed — masks - Naval ranks get naval masks: `капітан 1-го рангу` and `капітан-лейтенант` were masked as army ranks (`майор`, `лейтенант`). diff --git a/datamasking/_textio.py b/datamasking/_textio.py index 754ec6d..715af35 100644 --- a/datamasking/_textio.py +++ b/datamasking/_textio.py @@ -15,6 +15,9 @@ from typing import Iterable, List, Optional, Tuple DEFAULT_ENCODING = "utf-8" +# UTF-32 — перед UTF-16: BOM UTF-32-LE починається з BOM UTF-16-LE +_BOMS = ((codecs.BOM_UTF32_LE, "utf-32-le"), (codecs.BOM_UTF32_BE, "utf-32-be"), + (codecs.BOM_UTF16_LE, "utf-16-le"), (codecs.BOM_UTF16_BE, "utf-16-be")) DEFAULT_ALLOWED = ("utf-8", "cp1251", "latin-1") @@ -56,15 +59,32 @@ def read_text(path: Path, encoding: str = DEFAULT_ENCODING, Переноси рядків не перетворюються (як open(..., newline='')). """ raw = Path(path).read_bytes() + if encoding == "auto": + # BOM однозначно називає кодування. До 3.1.21 UTF-16 «декодувався» + # як cp1251 (однобайтові кодування приймають будь-які байти), і + # документ виходив незамаскованим з кодом 0. Кодек із явним порядком + # байтів лишає BOM у тексті символом U+FEFF — і вихід пишеться так само + for bom, enc in _BOMS: + if raw.startswith(bom): + return raw.decode(enc), enc candidates = list(allowed) if encoding == "auto" else [encoding] last_error: Exception = UnicodeDecodeError(DEFAULT_ENCODING, b"", 0, 1, "no encodings to try") for enc in candidates: try: - return raw.decode(enc), normalize_encoding(enc) + text = raw.decode(enc) except UnicodeDecodeError as e: last_error = e + continue + if encoding == "auto" and "\x00" in text: + # Символи NUL у «тексті» — це UTF-16/32 без BOM, а не cp1251 + pos = text.index("\x00") + last_error = UnicodeDecodeError(enc, raw, pos, pos + 1, + "NUL character (UTF-16 or UTF-32 without a BOM?)") + continue + return text, normalize_encoding(enc) if encoding == "auto": raise UnicodeDecodeError( getattr(last_error, "encoding", "auto"), raw, 0, 1, - f"none of the allowed encodings ({', '.join(candidates)}) could decode the file") + f"none of the allowed encodings ({', '.join(candidates)}) could decode the file " + f"({getattr(last_error, 'reason', last_error)})") raise last_error diff --git a/datamasking/extras/security.py b/datamasking/extras/security.py index 1fc5f83..bbbabb7 100644 --- a/datamasking/extras/security.py +++ b/datamasking/extras/security.py @@ -130,8 +130,11 @@ def _derive_key_v2(cls, password: str, salt: bytes, header: Dict[str, Any]) -> b r=int(header["r"]), p=int(header["p"])) return kdf.derive(password.encode("utf-8")) if header.get("kdf") == "pbkdf2": - iterations = int(header.get("iterations", cls.ITERATIONS)) - if not 100_000 <= iterations <= 10_000_000: + # Заголовок читається з файлу: [1] чи 1e400 до 3.1.21 давали + # TypeError/OverflowError з traceback замість помилки файлу + iterations = header.get("iterations", cls.ITERATIONS) + if isinstance(iterations, bool) or not isinstance(iterations, int) \ + or not 100_000 <= iterations <= 10_000_000: raise ValueError("Encrypted file header has an invalid PBKDF2 iteration count") return PBKDF2HMAC(algorithm=hashes.SHA256(), length=cls.KEY_LENGTH, salt=salt, iterations=iterations).derive(password.encode("utf-8")) @@ -262,7 +265,12 @@ def _decrypt_v2(self, raw: bytes, password: str) -> Dict[str, Any]: nonce_end = salt_end + self.NONCE_LENGTH if len(raw) < nonce_end + 16: raise ValueError("Encrypted file is too short.") - key = self._derive_key_v2(password, raw[prefix_end:salt_end], header) + # Параметри — перевірені check_kdf, а не сирий заголовок (у ньому могло + # не бути «n» для scrypt: KeyError з traceback до 3.1.21) + kdf_params: Dict[str, Any] = dict(params) + if "iterations" in header: + kdf_params["iterations"] = header["iterations"] + key = self._derive_key_v2(password, raw[prefix_end:salt_end], kdf_params) try: plaintext = AESGCM(key).decrypt(raw[salt_end:nonce_end], raw[nonce_end:], raw[:prefix_end]) except Exception as exc: diff --git a/datamasking/extras/selective.py b/datamasking/extras/selective.py index 3548c1a..a774996 100644 --- a/datamasking/extras/selective.py +++ b/datamasking/extras/selective.py @@ -337,6 +337,11 @@ def apply_filter_to_globals( # Тип «date» охоплює і цифрові, і текстові дати if "MASK_DATES" in globals_dict: globals_dict["MASK_DATE_TEXT"] = globals_dict["MASK_DATES"] + # «order_number» — і номери інших документів («довідка № 4567»): до + # 3.1.21 --only surname їх усе одно маскувало, а --exclude order_number + # не вимикало + if "MASK_ORDERS" in globals_dict: + globals_dict["MASK_DOCUMENT_NUMBERS"] = globals_dict["MASK_ORDERS"] # ============================================================================ diff --git a/datamasking/masking/cli.py b/datamasking/masking/cli.py index b37b868..8f8b249 100644 --- a/datamasking/masking/cli.py +++ b/datamasking/masking/cli.py @@ -846,6 +846,10 @@ def _validate_output_paths(args, input_path: Path, output_path: Path) -> Optiona перезаписує існуючий файл без --force. Повертає текст помилки або None.""" try: same = output_path.resolve() == input_path.resolve() + # Жорстке посилання на вхід — той самий файл (до 3.1.21 «-o» на + # нього з --force перезаписував оригінал) + if not same and output_path.exists(): + same = os.path.samefile(output_path, input_path) except OSError: same = False if same: @@ -856,6 +860,21 @@ def _validate_output_paths(args, input_path: Path, output_path: Path) -> Optiona def _resolve_password(args, config, logger) -> Tuple[Optional[str], Optional[str]]: + """Пароль для --encrypt (див. _resolve_password_impl) — і перевірка, що + його можна закодувати: пароль із байтами не в UTF-8 (з оболонки в іншому + кодуванні) до 3.1.21 виявлявся лише під час шифрування mapping, коли + замаскований файл уже був записаний, а mapping — ні.""" + password, error = _resolve_password_impl(args, config, logger) + if password is not None: + try: + password.encode("utf-8") + except UnicodeEncodeError: + return None, ("the password is not valid UTF-8 text (check the terminal or " + "environment encoding)") + return password, error + + +def _resolve_password_impl(args, config, logger) -> Tuple[Optional[str], Optional[str]]: """Визначає пароль для --encrypt ДО запису будь-яких файлів. Пріоритет: --password → --password-env VAR → $DATA_MASKING_PASSWORD → @@ -920,7 +939,13 @@ def _read_input(input_path: Path, is_json: bool, logger, masking_dict: Optional[ masking_dict["input_encoding"] = used if logger: logger.info(f"Input encoding: {used}") - return json.loads(text) if is_json else text + if masking_dict is not None: + # unmask читає вихід як JSON чи текст за цим полем, а не за + # суфіксом файлу (до 3.1.21 «-o out.json» для тексту не + # розмасковувався, а JSON у «.txt» відновлювався частково) + masking_dict["input_format"] = "json" if is_json else "text" + # BOM перед JSON (Блокнот Windows) — не помилка, як і для тексту + return json.loads(text[1:] if text.startswith("\ufeff") else text) if is_json else text except (FileNotFoundError, PermissionError, OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e: print(f"Error reading file: {e}") @@ -1188,7 +1213,9 @@ def _save_results(masked_data, is_json: bool, masking_dict: Dict, total_unique: int, args, config, logger, chain=None, password: Optional[str] = None) -> int: """Save masked output, mapping (or chain), and report. Returns exit code.""" - re_mask_passes = getattr(args, 're_mask', None) + # У звіті — виконані проходи, а не запитані (--re-mask 50 обмежується + # remask.max_passes; до 3.1.21 звіт показував 50) + re_mask_passes = len(chain.passes) if chain is not None else None system_cfg = getattr(config, 'system', None) try: @@ -1204,6 +1231,8 @@ def _save_results(masked_data, is_json: bool, masking_dict: Dict, # Кодуємо ДО відкриття файлу: інакше помилка кодування (символ поза # cp1251) лишала порожній вихідний файл без mapping (v3.1.8) out_text = json.dumps(masked_data, ensure_ascii=False, indent=2) if is_json else masked_data + if is_json and out_encoding.startswith(("utf-16-", "utf-32-")): + out_text = "\ufeff" + out_text # вхід мав BOM (див. _textio.read_text) payload = out_text.encode(out_encoding) with open(output_path, 'wb') as f: f.write(payload) @@ -1211,8 +1240,9 @@ def _save_results(masked_data, is_json: bool, masking_dict: Dict, if chain is not None: chain_json = map_path.with_name(map_path.name.replace("masking_map_", "masking_chain_", 1)) chain_dict = chain.to_dict() - if "input_encoding" in masking_dict: - chain_dict["input_encoding"] = masking_dict["input_encoding"] + for key in ("input_encoding", "input_format"): + if key in masking_dict: + chain_dict[key] = masking_dict[key] written_map = _write_mapping(chain_dict, chain_json, password, logger, kdf=getattr(args, '_kdf', None)) print(f" Chain mapping ({len(chain.passes)} passes): {written_map}") @@ -1269,7 +1299,12 @@ def main(argv=None) -> int: if Path(target).exists() and not args.force: print(f"Error: {target} already exists (use --force to overwrite)") return EXIT_ERROR - generate_default_config(target) + try: + generate_default_config(target) + except OSError as e: + # Немає теки, -c вказує на каталог, немає прав (до 3.1.21 — traceback) + print(f"Error: cannot write {target}: {e.strerror or e}") + return EXIT_ERROR print(f"Generated {target}") try: import yaml # noqa: F401 @@ -1282,11 +1317,6 @@ def main(argv=None) -> int: if getattr(args, 'password', None) is not None and getattr(args, 'password_env', None): print("Error: use either --password or --password-env, not both") return EXIT_USAGE - if (getattr(args, 'password', None) is not None or getattr(args, 'password_env', None)) \ - and not getattr(args, 'encrypt', False): - print("Error: --password / --password-env require --encrypt (the mapping would be written unencrypted)") - return EXIT_USAGE - # ================================================================ # Handle --list-types (early exit) # ================================================================ @@ -1319,6 +1349,13 @@ def main(argv=None) -> int: if settings_error: print(f"Error: {settings_error}") return EXIT_ERROR + # Після конфігурації: security.encrypt_output: true — те саме, що + # --encrypt (до 3.1.21 перевірка стояла раніше, і --password з таким + # конфігом відхилявся) + if (getattr(args, 'password', None) is not None or getattr(args, 'password_env', None)) \ + and not getattr(args, 'encrypt', False): + print("Error: --password / --password-env require --encrypt (the mapping would be written unencrypted)") + return EXIT_USAGE for strict_error in getattr(args, '_strict_errors', []): print(f"Error: strict_mode: {strict_error}") if getattr(args, '_strict_errors', []): diff --git a/datamasking/unmasking/cli.py b/datamasking/unmasking/cli.py index 5b8522a..1b93c55 100644 --- a/datamasking/unmasking/cli.py +++ b/datamasking/unmasking/cli.py @@ -95,15 +95,21 @@ def _unmask_encoding(config, masking_map) -> Tuple[str, List[str]]: def _size_limit(config: Any) -> int: - """Ліміт розміру файлу: менший із system.max_file_size_mb і - validation.max_input_size_mb (до 3.1.10 unmask їх не читав).""" + """Ліміт розміру файлу — як у mask: validation.max_input_size_mb (або + 100 МБ), але не більше за system.max_file_size_mb (до 3.1.10 unmask їх + не читав, до 3.1.21 не дозволяв підняти ліміт понад 100 МБ, хоча mask + дозволяв — і завеликий для unmask файл уже був замаскований).""" from datamasking.unmasking.helpers import MAX_INPUT_FILE_SIZE - limit = MAX_INPUT_FILE_SIZE - for section, key in (("system", "max_file_size_mb"), ("validation", "max_input_size_mb")): + + def mb(section: str, key: str) -> Optional[int]: value = getattr(getattr(config, section, None), key, None) if isinstance(value, int) and not isinstance(value, bool) and value > 0: - limit = min(limit, value * 1024 * 1024) - return limit + return value * 1024 * 1024 + return None + + limit = mb("validation", "max_input_size_mb") or MAX_INPUT_FILE_SIZE + system_limit = mb("system", "max_file_size_mb") + return min(limit, system_limit) if system_limit else limit def _expected_instances(masking_map: Dict) -> int: @@ -281,7 +287,7 @@ def log_debug(msg): if not password: print(f"❌ Змінна оточення '{args.password_env}' не встановлена або порожня") log_error(f"Environment variable '{args.password_env}' is not set or empty") - return EXIT_ERROR + return EXIT_USAGE # як у mask (до 3.1.21 тут був 1) else: password = _config_password(config) # security.password_env_var / password_generation.env_var — назва @@ -328,7 +334,7 @@ def log_debug(msg): else: print("❌ Будь ласка, вкажіть файл маппінгу через --map") log_error("Файл маппінгу не вказано") - return EXIT_ERROR + return EXIT_USAGE if not map_path.exists(): print(f"❌ Файл маппінгу не знайдено: {map_path}") @@ -379,18 +385,25 @@ def log_debug(msg): if not is_chain_mapping(masking_map): print("❌ --to-version працює лише з chain mapping (--re-mask)") log_error("--to-version on a non-chain mapping") - return EXIT_ERROR + return EXIT_USAGE if to_version < 0 or to_version > len(masking_map.get("passes", [])): + # Неправильне використання — код 2, як і всюди (до 3.1.21 — 1) print(f"❌ --to-version має бути в межах 0..{len(masking_map.get('passes', []))}") - return EXIT_ERROR + return EXIT_USAGE validate_file_size(masked_path, max_size=_size_limit(config)) # Кодування: з mapping (mask записує його, якщо вхід був не в utf-8); # system.encoding у конфігурації, якщо задано явно, має пріоритет file_encoding, allowed = _unmask_encoding(config, masking_map) masked_text, file_encoding = read_text(masked_path, file_encoding, allowed) - if masked_path.suffix.lower() == '.json': - masked_data = json.loads(masked_text) + # JSON чи текст — як при маскуванні (input_format у mapping, з + # 3.1.21); у старих mapping цього поля немає — за суфіксом файлу + if isinstance(masking_map, dict) and masking_map.get("input_format") in ("json", "text"): + is_json = masking_map["input_format"] == "json" + else: + is_json = masked_path.suffix.lower() == '.json' + if is_json: + masked_data = json.loads(masked_text[1:] if masked_text.startswith("\ufeff") else masked_text) else: masked_data = masked_text @@ -415,7 +428,7 @@ def log_debug(msg): print(f"🔄 Розмаскування {masked_path.name} (ланцюг з {total_passes} проходів)...") log_info(f"Розмаскування ланцюга з {total_passes} проходів") - if masked_path.suffix.lower() == '.json': + if is_json: restored_data, stats = unmask_json_chain_with_stats( masked_data, masking_map, to_version=getattr(args, 'to_version', None) or 0) else: @@ -426,7 +439,7 @@ def log_debug(msg): print(f"🔄 Розмаскування {masked_path.name} (логіка {map_version})...") log_info(f"Розмаскування {masked_path.name} (логіка {map_version})") - if masked_path.suffix.lower() == '.json': + if is_json: restored_data, stats = unmask_json_with_stats(masked_data, masking_map, map_version) else: if map_version.startswith("v2"): @@ -440,8 +453,9 @@ def log_debug(msg): try: # Відновлений текст чутливіший за mapping: атомарно, 0600 (v3.1.10) - out_text = json.dumps(restored_data, ensure_ascii=False, indent=2) \ - if masked_path.suffix.lower() == '.json' else restored_data + out_text = json.dumps(restored_data, ensure_ascii=False, indent=2) if is_json else restored_data + if is_json and file_encoding.startswith(("utf-16-", "utf-32-")): + out_text = "\ufeff" + out_text atomic_write_private(output_path, out_text.encode(file_encoding)) elapsed = time.time() - start_time diff --git a/datamasking/unmasking/io.py b/datamasking/unmasking/io.py index 10c0d7b..a1a9dc5 100644 --- a/datamasking/unmasking/io.py +++ b/datamasking/unmasking/io.py @@ -115,12 +115,16 @@ def _validate_mappings(mapping: Dict, where: str) -> None: if not isinstance(entries, dict): raise ValueError(f"'{where}mappings.{category}' must be a dictionary") for original, info in entries.items(): + # Порожня маска «знаходиться» між будь-якими двома символами, і + # unmask вставляв оригінал у випадкові місця (до 3.1.21) if isinstance(info, str): + if not info: + raise ValueError(f"'{where}mappings.{category}.{original}' must not be empty") continue if not isinstance(info, dict): raise ValueError(f"'{where}mappings.{category}.{original}' must be an object or a string") - if not isinstance(info.get("masked_as"), str): - raise ValueError(f"'{where}mappings.{category}.{original}.masked_as' must be a string") + if not isinstance(info.get("masked_as"), str) or not info["masked_as"]: + raise ValueError(f"'{where}mappings.{category}.{original}.masked_as' must be a non-empty string") instances = info.get("instances", []) if not isinstance(instances, list) or not all(isinstance(n, int) for n in instances): raise ValueError(f"'{where}mappings.{category}.{original}.instances' must be a list of integers") diff --git a/tests/test_cli_findings.py b/tests/test_cli_findings.py new file mode 100644 index 0000000..8033fad --- /dev/null +++ b/tests/test_cli_findings.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Знахідки аудиту CLI та вводу-виводу (після 3.1.20). + + - system.encoding: auto читав UTF-16 як cp1251 — нічого не маскувалось, + exit 0; + - --only / --exclude не керували номерами документів («довідка № 4567»); + - unmask вибирав JSON чи текст за суфіксом замаскованого файлу, а mask — + за суфіксом вхідного; JSON з BOM відхилявся; + - unmask не дозволяв підняти ліміт розміру понад 100 МБ; + - security.encrypt_output: true + --password відхилялось; + - пошкоджений заголовок .enc формату 2, --init-config у неіснуючу теку — + traceback; + - пароль не в UTF-8 лишав замаскований файл без mapping; + - звіт показував запитану кількість проходів --re-mask, а не виконану; + - «-o» — жорстке посилання на вхід — перезаписувало оригінал; + - порожній masked_as вставляв оригінал між усіма символами; + - коди виходу unmask для неправильного використання — 1 замість 2. +""" +import codecs +import json +import os +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from datamasking import _textio # noqa: E402 +from datamasking.masking import cli as mask_cli # noqa: E402 +from datamasking.unmasking import cli as unmask_cli # noqa: E402 +from datamasking.unmasking.io import validate_mapping_schema # noqa: E402 + +try: + import cryptography # noqa: F401 + _HAS_CRYPTO = True +except ImportError: + _HAS_CRYPTO = False +needs_crypto = pytest.mark.skipif(not _HAS_CRYPTO, reason="cryptography not installed") + +DOC = ("Сидоренко Олександр Петрович, ІПН 1234567890, народився 12.05.1990.\r\n" + "Старший сержант Коваленко Іван Іванович, в/ч А1234.\r\n") + + +@pytest.fixture +def cwd(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + for var in [v for v in os.environ if v.startswith("DATA_MASKING_")]: + monkeypatch.delenv(var, raising=False) + return tmp_path + + +def _config(path: Path, data) -> None: + path.joinpath("config.py").write_text(f"CONFIG = {data!r}\n", encoding="utf-8") + + +def _map(path: Path) -> str: + return str(next(path.glob("masking_map_*"))) + + +class TestEncodingAuto: + @pytest.mark.parametrize("codec", ["utf-16-le", "utf-16-be", "utf-32-le", "utf-32-be"]) + def test_bom_round_trip(self, cwd, codec): + bom = "\ufeff".encode(codec) + _config(cwd, {"system": {"encoding": "auto"}}) + raw = bom + DOC.encode(codec) + (cwd / "in.txt").write_bytes(raw) + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--no-report"]) == 0 + masked = (cwd / "out.txt").read_bytes() + assert masked.startswith(bom) + assert "Сидоренко" not in masked.decode(codec) + assert unmask_cli.main(["out.txt", "--map", _map(cwd), "-o", "rec.txt"]) == 0 + assert (cwd / "rec.txt").read_bytes() == raw + + def test_nul_is_not_cp1251(self, tmp_path): + path = tmp_path / "in.txt" + path.write_bytes(DOC.encode("utf-16-le")) # без BOM + with pytest.raises(UnicodeDecodeError, match="NUL"): + _textio.read_text(path, "auto", ("utf-8", "cp1251", "latin-1")) + + def test_plain_files_unchanged(self, tmp_path): + path = tmp_path / "in.txt" + path.write_bytes(DOC.encode("cp1251")) + assert _textio.read_text(path, "auto", ("utf-8", "cp1251"))[1] == "cp1251" + path.write_bytes(codecs.BOM_UTF8 + DOC.encode("utf-8")) + assert _textio.read_text(path, "auto", ("utf-8", "cp1251")) == ("" + DOC, "utf-8") + + +class TestSelectiveDocumentNumbers: + TEXT = "Видано довідку № 4567 солдату Коваленку Івану Івановичу.\nРапорт № 891 подано.\n" + + @pytest.mark.parametrize("flags", [("--only", "surname"), ("--exclude", "order_number")]) + def test_document_numbers_follow_orders(self, cwd, flags): + (cwd / "in.txt").write_text(self.TEXT, encoding="utf-8") + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--no-report", *flags]) == 0 + out = (cwd / "out.txt").read_text(encoding="utf-8") + assert "№ 4567" in out and "№ 891" in out and "Коваленку" not in out + + +class TestJsonFormat: + def test_text_written_as_json_name(self, cwd): + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + assert mask_cli.main(["-i", "in.txt", "-o", "out.json", "--no-report"]) == 0 + assert unmask_cli.main(["out.json", "--map", _map(cwd), "-o", "rec.txt"]) == 0 + assert (cwd / "rec.txt").read_bytes() == DOC.encode("utf-8") + + def test_json_written_as_txt_name(self, cwd): + data = {"a": "Перший рядок\nСтарший сержант Коваленко Іван Іванович"} + (cwd / "src.json").write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + assert mask_cli.main(["-i", "src.json", "-o", "masked.txt", "--no-report"]) == 0 + assert unmask_cli.main(["masked.txt", "--map", _map(cwd), "-o", "rec.txt"]) == 0 + assert json.loads((cwd / "rec.txt").read_text(encoding="utf-8")) == data + + def test_json_with_bom(self, cwd): + data = {"a": "солдат Петренко Іван Іванович"} + (cwd / "in.json").write_bytes(codecs.BOM_UTF8 + json.dumps(data, ensure_ascii=False).encode("utf-8")) + assert mask_cli.main(["-i", "in.json", "-o", "out.json", "--no-report"]) == 0 + assert "Петренко" not in (cwd / "out.json").read_text(encoding="utf-8") + assert unmask_cli.main(["out.json", "--map", _map(cwd), "-o", "rec.json"]) == 0 + assert json.loads((cwd / "rec.json").read_text(encoding="utf-8")) == data + + +class TestLimitsAndPaths: + def test_unmask_size_limit_follows_validation(self): + class Section: + def __init__(self, **kw): + self.__dict__.update(kw) + + class Cfg: + validation = Section(max_input_size_mb=200) + system = Section(max_file_size_mb=None) + assert unmask_cli._size_limit(Cfg()) == 200 * 1024 * 1024 + Cfg.system = Section(max_file_size_mb=150) + assert unmask_cli._size_limit(Cfg()) == 150 * 1024 * 1024 + + @needs_crypto + def test_encrypt_output_with_password(self, cwd): + _config(cwd, {"security": {"encrypt_output": True}}) + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--no-report", "--password", "Pw-123456"]) == 0 + assert list(cwd.glob("masking_map_*.enc")) and not list(cwd.glob("masking_map_*.json")) + + def test_password_without_encrypt_still_usage_error(self, cwd): + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--password", "x"]) == 2 + + @needs_crypto + def test_non_utf8_password(self, cwd): + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--encrypt", + "--password", "pa\udcffss"]) == 2 + assert not (cwd / "out.txt").exists() + + @pytest.mark.parametrize("target", ["nodir/c.yaml", "."]) + def test_init_config_bad_target(self, cwd, capsys, target): + assert mask_cli.main(["--init-config", "-c", target, "--force"]) == 1 + assert "cannot write" in capsys.readouterr().out + + def test_report_shows_executed_passes(self, cwd): + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + assert mask_cli.main(["-i", "in.txt", "-o", "out.txt", "--re-mask", "50"]) == 0 + report = next(cwd.glob("masking_report_*.txt")).read_text(encoding="utf-8") + assert "(10 проходів)" in report + + @pytest.mark.skipif(not hasattr(os, "link"), reason="no hard links") + def test_hard_link_to_input(self, cwd): + (cwd / "in.txt").write_bytes(DOC.encode("utf-8")) + os.link(cwd / "in.txt", cwd / "link.txt") + assert mask_cli.main(["-i", "in.txt", "-o", "link.txt", "--force", "--no-report"]) == 2 + assert (cwd / "in.txt").read_bytes() == DOC.encode("utf-8") + + +class TestUnmaskInputs: + @pytest.mark.parametrize("mapping", [ + {"version": "2.1", "mappings": {"surname": {"Коваленко": {"masked_as": "", "instances": [1]}}}}, + {"mappings": {"surname": {"Коваленко": ""}}}, + ]) + def test_empty_mask_rejected(self, mapping): + with pytest.raises(ValueError, match="empty"): + validate_mapping_schema(mapping) + + def test_usage_exit_codes(self, cwd, monkeypatch): + (cwd / "masked.txt").write_text("x", encoding="utf-8") + assert unmask_cli.main(["masked.txt"]) == 2 + (cwd / "m.json").write_text('{"version": "2.1", "mappings": {}}', encoding="utf-8") + monkeypatch.delenv("NO_SUCH_VAR", raising=False) + assert unmask_cli.main(["masked.txt", "--map", "m.json", "--password-env", "NO_SUCH_VAR"]) == 2 + + +class TestEncHeader: + @needs_crypto + @pytest.mark.parametrize("header", [ + {"v": 2, "kdf": "scrypt", "salt_length": 16}, + {"v": 2, "kdf": "pbkdf2", "salt_length": 16, "iterations": [1]}, + {"v": 2, "kdf": "pbkdf2", "salt_length": 16, "iterations": 1e300}, + ]) + def test_crafted_header_is_a_file_error(self, cwd, capsys, header): + hb = json.dumps(header).encode("ascii") + (cwd / "m.enc").write_bytes(b"DMENC2" + len(hb).to_bytes(2, "big") + hb + os.urandom(16 + 12 + 64)) + (cwd / "masked.txt").write_text("x", encoding="utf-8") + assert unmask_cli.main(["masked.txt", "--map", "m.enc", "--password", "pw", "-o", "r.txt"]) == 1 + assert "Traceback" not in capsys.readouterr().err diff --git a/tests/test_to_version.py b/tests/test_to_version.py index c2efc37..56c97bd 100644 --- a/tests/test_to_version.py +++ b/tests/test_to_version.py @@ -75,10 +75,10 @@ def test_cli_partial_and_full(self, chain_run): def test_cli_rejects_non_chain_and_out_of_range(self, chain_run): tmp_path, chain_path, _, _ = chain_run - assert unmask_cli.main(["m.txt", "--map", str(chain_path), "--to-version", "9", "-o", "x.txt"]) == 1 + assert unmask_cli.main(["m.txt", "--map", str(chain_path), "--to-version", "9", "-o", "x.txt"]) == 2 assert mask_cli.main(["-i", "in.txt", "-o", "s.txt", "--no-report"]) == 0 smap = next(tmp_path.glob("masking_map_*.json")) - assert unmask_cli.main(["s.txt", "--map", str(smap), "--to-version", "0", "-o", "y.txt"]) == 1 + assert unmask_cli.main(["s.txt", "--map", str(smap), "--to-version", "0", "-o", "y.txt"]) == 2 class TestChainUnmaskerApi: