diff --git a/CHANGELOG.md b/CHANGELOG.md index 095b614..63bd0e5 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,109 @@ 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. + +### 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 `капітан-лейтенант` 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/_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/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/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 44df996..8f8b249 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) @@ -816,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: @@ -826,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 → @@ -890,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}") @@ -948,6 +1003,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 +1025,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) і @@ -1149,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: @@ -1165,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) @@ -1172,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}") @@ -1230,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 @@ -1243,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) # ================================================================ @@ -1280,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/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/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/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/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_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 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: