diff --git a/breach-check/.gitignore b/breach-check/.gitignore new file mode 100644 index 0000000..b1b2c79 --- /dev/null +++ b/breach-check/.gitignore @@ -0,0 +1,6 @@ +.venv/ +__pycache__/ +*.egg-info/ +.pytest_cache/ +.mypy_cache/ +.env diff --git a/breach-check/pyproject.toml b/breach-check/pyproject.toml new file mode 100644 index 0000000..d7d58b5 --- /dev/null +++ b/breach-check/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "breach-check" +version = "0.1.0" +description = "Проверка паролей и email по Have I Been Pwned: пароли через k-anonymity range API. Zero runtime dependencies." +requires-python = ">=3.11" +authors = [{ name = "h4root" }] +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest>=8.0"] + +[project.scripts] +breach-check = "breach_check.cli:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/breach_check"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/breach-check/src/breach_check/__init__.py b/breach-check/src/breach_check/__init__.py new file mode 100644 index 0000000..7a1d935 --- /dev/null +++ b/breach-check/src/breach_check/__init__.py @@ -0,0 +1,17 @@ +from breach_check.accounts import AccountResult, Breach, check_account, read_accounts +from breach_check.hibp import HibpError +from breach_check.passwords import PasswordResult, check_password, parse_range, sha1_hex + +__all__ = [ + "AccountResult", + "Breach", + "HibpError", + "PasswordResult", + "check_account", + "check_password", + "parse_range", + "read_accounts", + "sha1_hex", + "__version__", +] +__version__ = "0.1.0" diff --git a/breach-check/src/breach_check/__main__.py b/breach-check/src/breach_check/__main__.py new file mode 100644 index 0000000..fd994a2 --- /dev/null +++ b/breach-check/src/breach_check/__main__.py @@ -0,0 +1,6 @@ +import sys + +from breach_check.cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/breach-check/src/breach_check/accounts.py b/breach-check/src/breach_check/accounts.py new file mode 100644 index 0000000..70add3e --- /dev/null +++ b/breach-check/src/breach_check/accounts.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import json +import time +from dataclasses import dataclass +from urllib.parse import quote + +from breach_check.apikey import API_KEY_ENV +from breach_check.hibp import DEFAULT_TIMEOUT, HibpError, http_get + +ACCOUNT_URL = "https://haveibeenpwned.com/api/v3/breachedaccount/" +DEFAULT_DELAY = 1.6 +MAX_RETRY_WAIT = 60.0 + + +@dataclass(frozen=True) +class Breach: + name: str + date: str | None = None + data_classes: tuple[str, ...] = () + + +@dataclass(frozen=True) +class AccountResult: + account: str + breaches: tuple[Breach, ...] = () + error: str | None = None + + +def validate_account(raw: str) -> str: + account = raw.strip() + local, _, domain = account.partition("@") + if not local or not domain: + raise ValueError(f"не похоже на email: {raw!r}") + return account + + +def read_accounts(text: str) -> list[str]: + lines = (line.split("#", 1)[0].strip() for line in text.splitlines()) + return [validate_account(line) for line in lines if line] + + +def parse_breaches(body: str) -> tuple[Breach, ...]: + try: + payload = json.loads(body) + except json.JSONDecodeError as error: + raise HibpError(f"HIBP вернул не JSON: {error}") from error + return tuple( + Breach( + item.get("Name") or item.get("Title") or "?", + item.get("BreachDate"), + tuple(item.get("DataClasses") or ()), + ) + for item in payload + ) + + +def check_account( + account: str, + api_key: str, + timeout: float = DEFAULT_TIMEOUT, +) -> AccountResult: + url = f"{ACCOUNT_URL}{quote(account, safe='')}?truncateResponse=false" + headers = {"hibp-api-key": api_key} + try: + response = http_get(url, headers, timeout) + if response.status == 429: + time.sleep(min(response.retry_after or DEFAULT_DELAY, MAX_RETRY_WAIT)) + response = http_get(url, headers, timeout) + except OSError as error: + return AccountResult(account, error=str(error)) + + if response.status == 404: + return AccountResult(account) + if response.status == 200: + return AccountResult(account, parse_breaches(response.body)) + if response.status in (401, 403): + raise HibpError( + f"HIBP отклонил запрос (HTTP {response.status}): проверьте {API_KEY_ENV}" + ) + return AccountResult(account, error=f"HTTP {response.status}") diff --git a/breach-check/src/breach_check/apikey.py b/breach-check/src/breach_check/apikey.py new file mode 100644 index 0000000..e52f830 --- /dev/null +++ b/breach-check/src/breach_check/apikey.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import os +import sys +from getpass import getpass +from pathlib import Path + +API_KEY_ENV = "HIBP_API_KEY" +KEY_URL = "https://haveibeenpwned.com/API/Key" +PROMPT = "Ключ HIBP (ввод не отображается): " + + +class MissingApiKey(RuntimeError): + pass + + +def parse_key_file(text: str) -> str: + for raw in text.splitlines(): + line = raw.split("#", 1)[0].strip() + if not line: + continue + name, separator, value = line.partition("=") + if separator and name.strip().upper() != API_KEY_ENV: + continue + key = (value if separator else line).strip().strip("\"'") + if key: + return key + raise MissingApiKey(f"ключ не найден в файле, ожидалась строка вида {API_KEY_ENV}=...") + + +def resolve_api_key(key_file: Path | None = None, prompt: bool = True) -> str: + if key_file: + return parse_key_file(key_file.read_text(encoding="utf-8")) + + from_env = os.environ.get(API_KEY_ENV, "").strip() + if from_env: + return from_env + + if prompt and sys.stdin.isatty(): + key = getpass(PROMPT).strip() + if key: + return key + + raise MissingApiKey( + f"нужен ключ HIBP: {API_KEY_ENV}=..., --key-file или ввод с клавиатуры " + f"({KEY_URL})" + ) diff --git a/breach-check/src/breach_check/cli.py b/breach-check/src/breach_check/cli.py new file mode 100644 index 0000000..69bc279 --- /dev/null +++ b/breach-check/src/breach_check/cli.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import argparse +import json +import sys +import time +from collections.abc import Sequence +from getpass import getpass +from pathlib import Path + +from breach_check.accounts import ( + DEFAULT_DELAY, + AccountResult, + check_account, + read_accounts, + validate_account, +) +from breach_check.apikey import API_KEY_ENV, MissingApiKey, resolve_api_key +from breach_check.hibp import DEFAULT_TIMEOUT, HibpError +from breach_check.passwords import PasswordResult, check_password + + +def _plural(count: int, forms: tuple[str, str, str]) -> str: + if count % 10 == 1 and count % 100 != 11: + return forms[0] + if count % 10 in (2, 3, 4) and count % 100 not in (12, 13, 14): + return forms[1] + return forms[2] + + +def _format_accounts_text(results: list[AccountResult]) -> str: + width = max(len(r.account) for r in results) + lines = [] + for result in results: + if result.error: + tail = f"ошибка: {result.error}" + elif not result.breaches: + tail = "чисто" + else: + named = ", ".join( + f"{b.name} ({b.date})" if b.date else b.name for b in result.breaches + ) + word = _plural(len(result.breaches), ("утечка", "утечки", "утечек")) + tail = f"{len(result.breaches)} {word}: {named}" + lines.append(f"{result.account:<{width}} {tail}") + return "\n".join(lines) + + +def _format_accounts_json(results: list[AccountResult]) -> str: + payload = [ + { + "account": r.account, + "breaches": [ + {"name": b.name, "date": b.date, "data_classes": list(b.data_classes)} + for b in r.breaches + ], + "error": r.error, + } + for r in results + ] + return json.dumps(payload, indent=2, ensure_ascii=False) + + +def _format_passwords_text(results: list[PasswordResult]) -> str: + width = max(len(r.label) for r in results) + lines = [] + for result in results: + tail = ( + f"найден в утечках, совпадений: {result.count}" + if result.count + else "не найден" + ) + lines.append(f"{result.label:<{width}} {tail}") + return "\n".join(lines) + + +def _format_passwords_json(results: list[PasswordResult]) -> str: + payload = [{"label": r.label, "count": r.count} for r in results] + return json.dumps(payload, indent=2, ensure_ascii=False) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="breach-check", + description="Проверка по Have I Been Pwned: пароли через k-anonymity " + "range API (ключ не нужен), email через API v3 (нужен ключ).", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + email = subparsers.add_parser( + "email", + help=f"Проверить email по утечкам. Ключ: --key-file, {API_KEY_ENV} " + "или ввод с клавиатуры.", + ) + email.add_argument("accounts", nargs="*", help="Адреса: user@example.com.") + email.add_argument( + "--file", + "-f", + type=Path, + help="Файл со списком адресов, по одному в строке (# — комментарий).", + ) + email.add_argument( + "--key-file", + type=Path, + help=f"Файл с ключом HIBP: строка {API_KEY_ENV}=... или сам ключ. " + "Приоритетнее переменной окружения.", + ) + email.add_argument( + "--delay", + type=float, + default=DEFAULT_DELAY, + help="Пауза между запросами в секундах: у HIBP жёсткий лимит.", + ) + + password = subparsers.add_parser( + "password", + help="Проверить пароль: наружу уходят только первые 5 символов SHA-1.", + ) + password.add_argument( + "--stdin", + action="store_true", + help="Читать пароли из stdin, по одному в строке, вместо запроса ввода.", + ) + + for subparser in (email, password): + subparser.add_argument( + "--timeout", + type=float, + default=DEFAULT_TIMEOUT, + help="Таймаут запроса в секундах.", + ) + subparser.add_argument( + "--format", + choices=("text", "json"), + default="text", + help="Формат вывода.", + ) + return parser + + +def _collect_accounts(args: argparse.Namespace) -> list[str]: + accounts = [validate_account(raw) for raw in args.accounts] + if args.file: + accounts += read_accounts(args.file.read_text(encoding="utf-8")) + return accounts + + +def _run_email(args: argparse.Namespace) -> int: + if not args.accounts and not args.file: + print("breach-check: укажите адреса или --file", file=sys.stderr) + return 2 + if args.file and not args.file.is_file(): + print(f"breach-check: файл не найден: {args.file}", file=sys.stderr) + return 2 + + try: + accounts = _collect_accounts(args) + except (OSError, UnicodeDecodeError) as error: + print(f"breach-check: не удалось прочитать файл: {error}", file=sys.stderr) + return 2 + except ValueError as error: + print(f"breach-check: {error}", file=sys.stderr) + return 2 + + if not accounts: + print("breach-check: список адресов пуст", file=sys.stderr) + return 2 + + try: + api_key = resolve_api_key(args.key_file) + except MissingApiKey as error: + print(f"breach-check: {error}", file=sys.stderr) + return 2 + except (OSError, UnicodeDecodeError) as error: + print(f"breach-check: не удалось прочитать ключ: {error}", file=sys.stderr) + return 2 + + results = [] + for index, account in enumerate(accounts): + if index: + time.sleep(args.delay) + results.append(check_account(account, api_key, args.timeout)) + + if args.format == "json": + print(_format_accounts_json(results)) + else: + print(_format_accounts_text(results)) + + failing = [r for r in results if r.breaches or r.error] + return 1 if failing else 0 + + +def _read_passwords(use_stdin: bool) -> list[tuple[str, str]]: + if use_stdin: + lines = sys.stdin.read().splitlines() + return [(f"строка {n}", line) for n, line in enumerate(lines, 1) if line] + password = getpass("Пароль (ввод не отображается): ") + return [("пароль", password)] if password else [] + + +def _run_password(args: argparse.Namespace) -> int: + entries = _read_passwords(args.stdin) + if not entries: + print("breach-check: пароль не введён", file=sys.stderr) + return 2 + + try: + results = [ + check_password(password, label, args.timeout) for label, password in entries + ] + except OSError as error: + print(f"breach-check: сеть недоступна: {error}", file=sys.stderr) + return 2 + + if args.format == "json": + print(_format_passwords_json(results)) + else: + print(_format_passwords_text(results)) + + return 1 if any(r.count for r in results) else 0 + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return _run_email(args) if args.command == "email" else _run_password(args) + except HibpError as error: + print(f"breach-check: {error}", file=sys.stderr) + return 2 diff --git a/breach-check/src/breach_check/hibp.py b/breach-check/src/breach_check/hibp.py new file mode 100644 index 0000000..450520f --- /dev/null +++ b/breach-check/src/breach_check/hibp.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from dataclasses import dataclass +from email.message import Message +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +USER_AGENT = "breach-check/0.1.0 (+https://github.com/h4root/webtools)" +DEFAULT_TIMEOUT = 10.0 + + +class HibpError(RuntimeError): + pass + + +@dataclass(frozen=True) +class Response: + status: int + body: str + retry_after: float | None = None + + +def _retry_after(headers: Message) -> float | None: + raw = headers.get("Retry-After") + if raw is None: + return None + try: + return float(raw) + except ValueError: + return None + + +def http_get( + url: str, + headers: dict[str, str] | None = None, + timeout: float = DEFAULT_TIMEOUT, +) -> Response: + request = Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})}) + try: + with urlopen(request, timeout=timeout) as response: + body = response.read().decode("utf-8", "replace") + return Response(response.status, body, _retry_after(response.headers)) + except HTTPError as error: + body = error.read().decode("utf-8", "replace") + return Response(error.code, body, _retry_after(error.headers)) diff --git a/breach-check/src/breach_check/passwords.py b/breach-check/src/breach_check/passwords.py new file mode 100644 index 0000000..71c718b --- /dev/null +++ b/breach-check/src/breach_check/passwords.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import hashlib +from dataclasses import dataclass + +from breach_check.hibp import DEFAULT_TIMEOUT, HibpError, http_get + +RANGE_URL = "https://api.pwnedpasswords.com/range/" +PREFIX_LENGTH = 5 + + +@dataclass(frozen=True) +class PasswordResult: + label: str + count: int + + +def sha1_hex(password: str) -> str: + return hashlib.sha1(password.encode("utf-8")).hexdigest().upper() + + +def split_hash(digest: str) -> tuple[str, str]: + return digest[:PREFIX_LENGTH], digest[PREFIX_LENGTH:] + + +def parse_range(body: str) -> dict[str, int]: + counts: dict[str, int] = {} + for line in body.splitlines(): + suffix, _, raw_count = line.strip().partition(":") + try: + count = int(raw_count.replace(",", "")) + except ValueError: + continue + if count > 0: # ответы с Add-Padding содержат фиктивные хеши с count == 0 + counts[suffix.upper()] = count + return counts + + +def fetch_range(prefix: str, timeout: float = DEFAULT_TIMEOUT) -> str: + response = http_get(RANGE_URL + prefix, {"Add-Padding": "true"}, timeout) + if response.status != 200: + raise HibpError(f"Pwned Passwords вернул HTTP {response.status}") + return response.body + + +def check_password( + password: str, + label: str, + timeout: float = DEFAULT_TIMEOUT, +) -> PasswordResult: + prefix, suffix = split_hash(sha1_hex(password)) + counts = parse_range(fetch_range(prefix, timeout)) + return PasswordResult(label, counts.get(suffix, 0)) diff --git a/breach-check/tests/test_accounts.py b/breach-check/tests/test_accounts.py new file mode 100644 index 0000000..d4208fa --- /dev/null +++ b/breach-check/tests/test_accounts.py @@ -0,0 +1,114 @@ +import json +import socket + +import pytest + +from breach_check import accounts +from breach_check.accounts import ( + check_account, + parse_breaches, + read_accounts, + validate_account, +) +from breach_check.hibp import HibpError, Response + +BREACH_BODY = json.dumps( + [ + { + "Name": "Adobe", + "BreachDate": "2013-10-04", + "DataClasses": ["Email addresses", "Passwords"], + }, + {"Name": "LinkedIn", "BreachDate": "2012-05-05", "DataClasses": []}, + ] +) + + +def test_validate_account_trims(): + assert validate_account(" user@example.com ") == "user@example.com" + + +@pytest.mark.parametrize("raw", ["", "user", "@example.com", "user@"]) +def test_validate_account_rejects_non_email(raw): + with pytest.raises(ValueError): + validate_account(raw) + + +def test_read_accounts_skips_comments_and_blanks(): + text = "# список\nuser@example.com\n\n admin@example.com # прод\n" + assert read_accounts(text) == ["user@example.com", "admin@example.com"] + + +def test_parse_breaches_extracts_fields(): + breaches = parse_breaches(BREACH_BODY) + assert [b.name for b in breaches] == ["Adobe", "LinkedIn"] + assert breaches[0].date == "2013-10-04" + assert breaches[0].data_classes == ("Email addresses", "Passwords") + + +def test_parse_breaches_rejects_garbage(): + with pytest.raises(HibpError): + parse_breaches("502") + + +def _stub_responses(monkeypatch, *responses): + calls: list[str] = [] + queue = list(responses) + + def get(url, headers=None, timeout=None): + calls.append(url) + result = queue.pop(0) + if isinstance(result, Exception): + raise result + return result + + monkeypatch.setattr(accounts, "http_get", get) + monkeypatch.setattr(accounts.time, "sleep", lambda seconds: None) + return calls + + +def test_breached_account_lists_breaches(monkeypatch): + calls = _stub_responses(monkeypatch, Response(200, BREACH_BODY)) + result = check_account("user@example.com", "key") + assert [b.name for b in result.breaches] == ["Adobe", "LinkedIn"] + assert result.error is None + assert "user%40example.com" in calls[0] + + +def test_clean_account_has_no_breaches(monkeypatch): + _stub_responses(monkeypatch, Response(404, "")) + result = check_account("user@example.com", "key") + assert result.breaches == () + assert result.error is None + + +def test_rate_limit_is_retried_once(monkeypatch): + calls = _stub_responses( + monkeypatch, Response(429, "", retry_after=2.0), Response(404, "") + ) + assert check_account("user@example.com", "key").error is None + assert len(calls) == 2 + + +def test_repeated_rate_limit_becomes_error(monkeypatch): + _stub_responses(monkeypatch, Response(429, ""), Response(429, "")) + assert check_account("user@example.com", "key").error == "HTTP 429" + + +@pytest.mark.parametrize("status", [401, 403]) +def test_bad_api_key_raises(monkeypatch, status): + _stub_responses(monkeypatch, Response(status, "")) + with pytest.raises(HibpError): + check_account("user@example.com", "key") + + +def test_server_error_becomes_error_field(monkeypatch): + _stub_responses(monkeypatch, Response(503, "")) + assert check_account("user@example.com", "key").error == "HTTP 503" + + +def test_network_failure_becomes_error_field(monkeypatch): + _stub_responses(monkeypatch, socket.gaierror("Name or service not known")) + result = check_account("user@example.com", "key") + assert result.breaches == () + assert "Name or service not known" in result.error diff --git a/breach-check/tests/test_apikey.py b/breach-check/tests/test_apikey.py new file mode 100644 index 0000000..1f32277 --- /dev/null +++ b/breach-check/tests/test_apikey.py @@ -0,0 +1,75 @@ +from pathlib import Path + +import pytest + +from breach_check import apikey +from breach_check.apikey import API_KEY_ENV, MissingApiKey, parse_key_file, resolve_api_key + + +class FakeStdin: + def __init__(self, interactive: bool): + self.interactive = interactive + + def isatty(self) -> bool: + return self.interactive + + +@pytest.fixture(autouse=True) +def no_ambient_key(monkeypatch): + monkeypatch.delenv(API_KEY_ENV, raising=False) + monkeypatch.setattr(apikey.sys, "stdin", FakeStdin(interactive=False)) + + +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("secret-key\n", "secret-key"), + (f"{API_KEY_ENV}=secret-key\n", "secret-key"), + (f'{API_KEY_ENV}="secret-key"\n', "secret-key"), + (f"# ключ\n\n {API_KEY_ENV} = secret-key \n", "secret-key"), + (f"OTHER=нет\n{API_KEY_ENV}=secret-key\n", "secret-key"), + ], +) +def test_parse_key_file(text, expected): + assert parse_key_file(text) == expected + + +@pytest.mark.parametrize("text", ["", "# только комментарий\n", "OTHER=нет\n", f"{API_KEY_ENV}=\n"]) +def test_parse_key_file_without_key_raises(text): + with pytest.raises(MissingApiKey): + parse_key_file(text) + + +def test_key_file_wins_over_environment(monkeypatch, tmp_path: Path): + monkeypatch.setenv(API_KEY_ENV, "from-env") + key_file = tmp_path / "hibp.key" + key_file.write_text("from-file\n", encoding="utf-8") + assert resolve_api_key(key_file) == "from-file" + + +def test_environment_is_used_without_key_file(monkeypatch): + monkeypatch.setenv(API_KEY_ENV, " from-env ") + assert resolve_api_key() == "from-env" + + +def test_prompt_is_used_when_terminal_is_interactive(monkeypatch): + monkeypatch.setattr(apikey.sys, "stdin", FakeStdin(interactive=True)) + monkeypatch.setattr(apikey, "getpass", lambda prompt: " typed-key ") + assert resolve_api_key() == "typed-key" + + +def test_empty_prompt_raises(monkeypatch): + monkeypatch.setattr(apikey.sys, "stdin", FakeStdin(interactive=True)) + monkeypatch.setattr(apikey, "getpass", lambda prompt: "") + with pytest.raises(MissingApiKey): + resolve_api_key() + + +def test_no_prompt_without_terminal(): + with pytest.raises(MissingApiKey): + resolve_api_key() + + +def test_missing_key_file_raises_oserror(tmp_path: Path): + with pytest.raises(OSError): + resolve_api_key(tmp_path / "нет.key") diff --git a/breach-check/tests/test_cli.py b/breach-check/tests/test_cli.py new file mode 100644 index 0000000..cf2893f --- /dev/null +++ b/breach-check/tests/test_cli.py @@ -0,0 +1,165 @@ +import io +import json +from pathlib import Path + +import pytest + +from breach_check import accounts, apikey, cli, passwords +from breach_check.apikey import API_KEY_ENV +from breach_check.cli import main +from breach_check.hibp import Response + +BREACH_BODY = json.dumps([{"Name": "Adobe", "BreachDate": "2013-10-04"}]) +PASSWORD_SUFFIX = "1E4C9B93F3F0682250B6CF8331B7EE68FD8" + + +@pytest.fixture +def api_key(monkeypatch): + monkeypatch.setenv(API_KEY_ENV, "test-key") + + +@pytest.fixture +def hibp(monkeypatch): + by_account: dict[str, object] = {} + + def get(url, headers=None, timeout=None): + for account, result in by_account.items(): + if account.replace("@", "%40") in url: + if isinstance(result, Exception): + raise result + return result + return Response(404, "") + + monkeypatch.setattr(accounts, "http_get", get) + return by_account + + +@pytest.fixture +def pwned(monkeypatch): + body = {"text": ""} + monkeypatch.setattr(passwords, "fetch_range", lambda prefix, timeout=None: body["text"]) + return body + + +def test_clean_email_exits_zero(api_key, hibp, capsys): + assert main(["email", "user@example.com", "--delay", "0"]) == 0 + assert "чисто" in capsys.readouterr().out + + +def test_breached_email_exits_one(api_key, hibp, capsys): + hibp["user@example.com"] = Response(200, BREACH_BODY) + assert main(["email", "user@example.com", "--delay", "0"]) == 1 + out = capsys.readouterr().out + assert "1 утечка" in out and "Adobe (2013-10-04)" in out + + +def test_missing_api_key_exits_two(monkeypatch, capsys): + monkeypatch.delenv(API_KEY_ENV, raising=False) + assert main(["email", "user@example.com"]) == 2 + assert API_KEY_ENV in capsys.readouterr().err + + +def test_key_file_is_used(monkeypatch, hibp, tmp_path: Path): + monkeypatch.delenv(API_KEY_ENV, raising=False) + key_file = tmp_path / "hibp.key" + key_file.write_text(f"{API_KEY_ENV}=test-key\n", encoding="utf-8") + assert main(["email", "user@example.com", "--key-file", str(key_file), "--delay", "0"]) == 0 + + +def test_missing_key_file_exits_two(monkeypatch, tmp_path: Path, capsys): + monkeypatch.delenv(API_KEY_ENV, raising=False) + args = ["email", "user@example.com", "--key-file", str(tmp_path / "нет.key")] + assert main([*args, "--delay", "0"]) == 2 + assert "ключ" in capsys.readouterr().err + + +def test_typed_key_is_used(monkeypatch, hibp, capsys): + monkeypatch.delenv(API_KEY_ENV, raising=False) + monkeypatch.setattr(apikey.sys, "stdin", type("S", (), {"isatty": lambda self: True})()) + monkeypatch.setattr(apikey, "getpass", lambda prompt: "typed-key") + assert main(["email", "user@example.com", "--delay", "0"]) == 0 + + +def test_bad_email_is_reported_before_key_is_requested(monkeypatch, capsys): + monkeypatch.delenv(API_KEY_ENV, raising=False) + monkeypatch.setattr(apikey, "getpass", lambda prompt: pytest.fail("ключ спрошен зря")) + assert main(["email", "not-an-email", "--delay", "0"]) == 2 + assert "не похоже на email" in capsys.readouterr().err + + +def test_bad_email_exits_two(api_key): + assert main(["email", "not-an-email", "--delay", "0"]) == 2 + + +def test_no_targets_exits_two(api_key): + assert main(["email", "--delay", "0"]) == 2 + + +def test_email_file_is_checked(api_key, hibp, tmp_path: Path, capsys): + hibp["b@example.com"] = Response(200, BREACH_BODY) + listing = tmp_path / "emails.txt" + listing.write_text("# люди\na@example.com\nb@example.com\n", encoding="utf-8") + assert main(["email", "--file", str(listing), "--delay", "0"]) == 1 + out = capsys.readouterr().out + assert "a@example.com" in out and "b@example.com" in out + + +def test_missing_email_file_exits_two(api_key, tmp_path: Path): + assert main(["email", "--file", str(tmp_path / "nope.txt"), "--delay", "0"]) == 2 + + +def test_email_network_error_exits_one(api_key, hibp, capsys): + hibp["user@example.com"] = OSError("connection reset") + assert main(["email", "user@example.com", "--delay", "0"]) == 1 + assert "ошибка" in capsys.readouterr().out + + +def test_email_json_output_is_parseable(api_key, hibp, capsys): + hibp["user@example.com"] = Response(200, BREACH_BODY) + main(["email", "user@example.com", "--delay", "0", "--format", "json"]) + payload = json.loads(capsys.readouterr().out) + assert payload[0]["account"] == "user@example.com" + assert payload[0]["breaches"][0]["name"] == "Adobe" + assert payload[0]["error"] is None + + +def test_pwned_password_from_stdin_exits_one(pwned, monkeypatch, capsys): + pwned["text"] = f"{PASSWORD_SUFFIX}:9659365\n" + monkeypatch.setattr("sys.stdin", io.StringIO("password\n")) + assert main(["password", "--stdin"]) == 1 + assert "совпадений: 9659365" in capsys.readouterr().out + + +def test_unknown_password_exits_zero(pwned, monkeypatch, capsys): + pwned["text"] = "0000000000000000000000000000000000A:7\n" + monkeypatch.setattr("sys.stdin", io.StringIO("password\n")) + assert main(["password", "--stdin"]) == 0 + assert "не найден" in capsys.readouterr().out + + +def test_password_is_never_printed(pwned, monkeypatch, capsys): + pwned["text"] = f"{PASSWORD_SUFFIX}:9659365\n" + monkeypatch.setattr("sys.stdin", io.StringIO("password\n")) + main(["password", "--stdin", "--format", "json"]) + captured = capsys.readouterr() + assert "password" not in captured.out + captured.err + + +def test_prompted_password_is_checked(pwned, monkeypatch, capsys): + pwned["text"] = f"{PASSWORD_SUFFIX}:9659365\n" + monkeypatch.setattr(cli, "getpass", lambda prompt: "password") + assert main(["password"]) == 1 + + +def test_empty_password_exits_two(monkeypatch): + monkeypatch.setattr(cli, "getpass", lambda prompt: "") + assert main(["password"]) == 2 + + +def test_password_json_output_is_parseable(pwned, monkeypatch, capsys): + pwned["text"] = f"{PASSWORD_SUFFIX}:9659365\n" + monkeypatch.setattr("sys.stdin", io.StringIO("password\nsecond\n")) + main(["password", "--stdin", "--format", "json"]) + payload = json.loads(capsys.readouterr().out) + assert [item["label"] for item in payload] == ["строка 1", "строка 2"] + assert payload[0]["count"] == 9659365 diff --git a/breach-check/tests/test_passwords.py b/breach-check/tests/test_passwords.py new file mode 100644 index 0000000..ff6cdb8 --- /dev/null +++ b/breach-check/tests/test_passwords.py @@ -0,0 +1,92 @@ +import pytest + +from breach_check import passwords +from breach_check.hibp import HibpError, Response +from breach_check.passwords import ( + check_password, + fetch_range, + parse_range, + sha1_hex, + split_hash, +) + +PASSWORD_HASH = "5BAA61E4C9B93F3F0682250B6CF8331B7EE68FD8" + + +def test_sha1_hex_is_uppercase_hex(): + assert sha1_hex("password") == PASSWORD_HASH + + +def test_sha1_hex_handles_non_ascii(): + assert len(sha1_hex("пароль")) == 40 + + +def test_split_hash_sends_only_five_characters(): + prefix, suffix = split_hash(PASSWORD_HASH) + assert prefix == "5BAA6" + assert len(suffix) == 35 + assert prefix + suffix == PASSWORD_HASH + + +def test_parse_range_reads_counts(): + body = "1E4C9B93F3F0682250B6CF8331B7EE68FD8:9659365\r\nAAAAA:12\r\n" + counts = parse_range(body) + assert counts["1E4C9B93F3F0682250B6CF8331B7EE68FD8"] == 9659365 + assert counts["AAAAA"] == 12 + + +def test_parse_range_drops_padding_entries(): + assert parse_range("ABC:0\nDEF:3\n") == {"DEF": 3} + + +def test_parse_range_ignores_malformed_lines(): + assert parse_range("\nне-строка\nDEF:3\nGHI:\n") == {"DEF": 3} + + +def test_parse_range_lowercase_suffix_is_normalised(): + assert parse_range("abc:5") == {"ABC": 5} + + +def _stub_range(monkeypatch, body: str) -> list[str]: + requested: list[str] = [] + + def fetch(prefix, timeout=None): + requested.append(prefix) + return body + + monkeypatch.setattr(passwords, "fetch_range", fetch) + return requested + + +def test_check_password_reports_count(monkeypatch): + requested = _stub_range(monkeypatch, f"{PASSWORD_HASH[5:]}:9659365\n") + result = check_password("password", "пароль") + assert (result.label, result.count) == ("пароль", 9659365) + assert requested == ["5BAA6"] + + +def test_check_password_not_found(monkeypatch): + _stub_range(monkeypatch, "0000000000000000000000000000000000A:7\n") + assert check_password("password", "пароль").count == 0 + + +def test_fetch_range_requests_padding(monkeypatch): + seen: dict[str, object] = {} + + def get(url, headers=None, timeout=None): + seen["url"] = url + seen["headers"] = headers + return Response(200, "ABC:1") + + monkeypatch.setattr(passwords, "http_get", get) + assert fetch_range("5BAA6") == "ABC:1" + assert seen["url"].endswith("/range/5BAA6") + assert seen["headers"]["Add-Padding"] == "true" + + +def test_fetch_range_raises_on_error_status(monkeypatch): + monkeypatch.setattr( + passwords, "http_get", lambda url, headers=None, timeout=None: Response(503, "") + ) + with pytest.raises(HibpError): + fetch_range("5BAA6")