Initial commit

This commit is contained in:
user
2026-07-12 14:06:45 +04:00
commit a75028d68f
148 changed files with 3275 additions and 0 deletions
+37
View File
@@ -0,0 +1,37 @@
# Python cache
__pycache__/
*.py[cod]
*$py.class
# Virtual environments
venv/
.venv/
env/
ENV/
# Environment variables
.env
# Build files
build/
dist/
*.egg-info/
# Jupyter Notebook
.ipynb_checkpoints/
# Test and coverage
.pytest_cache/
.coverage
htmlcov/
# IDE
.vscode/
.idea/
# OS files
.DS_Store
Thumbs.db
# Logs
*.log
+109
View File
@@ -0,0 +1,109 @@
from argparse import ArgumentParser
from secrets import randbelow
from utils.check_case import check_case
from utils.format_line import format_line
from orchestration.param_generator import ParamGenerator
def _random_secret() -> int:
# 2..9999 включительно (меньше 10000, исключаем 0/1)
return 2 + randbelow(9998)
def run_auto(count: int) -> None:
"""
Автоматический режим:
- случайно генерируем (p, q, g);
- случайно выбираем x, y (<10000);
- вычисляем K и проверяем через check_case;
- печатаем результат через format_line
"""
for i in range(1, count + 1):
p, q, g = ParamGenerator.generate() # q выводим для информации
x = _random_secret()
y = _random_secret()
# Вычислим K заранее, чтобы передать как эталон в check_case
# (тем самым используем общий хелпер единообразно)
ok, (A, B, K_a, K_b) = check_case(p, g, x, y, K_expected=None)
# Когда K_expected=None, наша старая версия check_case
# сравнивала строго с эталоном. Если у тебя именно так,
# просто заменим на повторный вызов с «эталоном = K_a»
if K_a != K_b:
ok = False
else:
ok, (A, B, K_a, K_b) = check_case(p, g, x, y, K_expected=K_a)
print(f"[auto #{i}] p={p} q={q} g={g}")
print(format_line(i, A, B, K_a, K_b, ok))
def run_manual(p: int, g: int, x: int, y: int, k: int | None) -> None:
"""
Ручной режим:
- принимаем p, g, x, y;
- если указан k — проверяем с эталоном;
- если k не указан — просто считаем и показываем
"""
if k is None:
ok, (A, B, K_a, K_b) = check_case(p, g, x, y, K_expected=None)
ok = (K_a == K_b)
print(f"[manual] p={p} g={g}")
print(format_line(1, A, B, K_a, K_b, ok))
else:
ok, (A, B, K_a, K_b) = check_case(p, g, x, y, K_expected=k)
print(f"[manual] p={p} g={g} K_expected={k}")
print(format_line(1, A, B, K_a, K_b, ok))
def parse_args():
ap = ArgumentParser(
description=(
"Demo DiffieHellman: режимы manual и auto\n"
"Auto: случайно генерирует (p,q,g) и секреты (<10000)\n"
"Manual: используется заданный набор параметров"
)
)
ap.add_argument(
"--mode",
choices=["auto", "manual"],
default="auto",
help="Режим работы (auto|manual). По умолчанию auto",
)
ap.add_argument(
"--count",
type=int,
default=5,
help="Сколько генераций в auto-режиме (по умолчанию 5)",
)
ap.add_argument("--p", type=int, help="Модуль (prime_modulus) в manual")
ap.add_argument("--g", type=int, help="Генератор (generator) в manual")
ap.add_argument("--x", type=int, help="Секрет Алисы (<10000) в manual")
ap.add_argument("--y", type=int, help="Секрет Боба (<10000) в manual")
ap.add_argument(
"--k",
type=int,
help="Эталонный общий ключ для проверки в manual (необязательно)",
)
return ap.parse_args()
def main():
args = parse_args()
if args.mode == "auto":
run_auto(count=args.count)
return
# manual
missing = [n for n, v in [("p", args.p), ("g", args.g),
("x", args.x), ("y", args.y)] if v is None]
if missing:
need = ", ".join(missing)
raise SystemExit(f"manual-режим: необходимо указать параметры: {need}")
run_manual(p=args.p, g=args.g, x=args.x, y=args.y, k=args.k)
if __name__ == "__main__":
main()
+1
View File
@@ -0,0 +1 @@
#
@@ -0,0 +1,19 @@
from secrets import randbelow
from .public_params import PublicParams
class Participant:
"""
Участник протокола (Алиса или Боб).
Держит свой секрет, публикует g^secret mod p,
умеет вычислять общий ключ.
"""
def __init__(self, params: PublicParams, secret: int | None = None):
self.params = params
self.secret = secret if secret is not None else (2 + randbelow(9998))
self.public_value = pow(self.params.generator,
self.secret, self.params.prime_modulus)
def compute_shared_key(self, other_public: int) -> int:
return pow(other_public, self.secret, self.params.prime_modulus)
@@ -0,0 +1,12 @@
def is_prime(n: int) -> bool:
"""Проверка на простоту"""
if n < 2:
return False
if n % 2 == 0:
return n == 2
d = 3
while d * d <= n:
if n % d == 0:
return False
d += 2
return True
@@ -0,0 +1,30 @@
from dataclasses import dataclass
from .prime_utils import is_prime
@dataclass(frozen=True)
class PublicParams:
"""
Публичные параметры протокола:
- prime_modulus: простое число p
- generator: порождающий элемент g
"""
prime_modulus: int
generator: int
def __post_init__(self):
p = self.prime_modulus
g = self.generator
# Проверка простоты модуля
if not is_prime(p):
raise ValueError(f"prime_modulus={p} не является простым числом")
# Проверка допустимости генератора
if g <= 1 or g >= p - 1:
raise ValueError(
f"generator={g} вне допустимого диапазона (2..p-2)")
# Проверка, что генератор не сворачивается в малую подгруппу
if pow(g, (p - 1) // 2, p) == 1:
raise ValueError(f"generator={g} не является корректным для p={p}")
@@ -0,0 +1,44 @@
from secrets import randbelow, randbits
from core.prime_utils import is_prime
class ParamGenerator:
"""
Генератор параметров протокола Диффи–Хеллмана.
Возвращает тройку (p, q, g):
- p = 2*q + 1, где p и q простые
- g — корректный генератор для группы по модулю p
"""
@staticmethod
def generate() -> tuple[int, int, int]:
"""
Случайным образом выбирает размерность (кол-во бит) для q,
подбирает безопасное простое p и генератор g.
Возвращает (p, q, g).
"""
# Размер числа выбираем случайно от 8 до 16 бит (от 256 до 65 535)
bits = 8 + randbelow(9)
# Генерация безопасного простого
while True:
q = randbits(bits - 1) | 1 # нечётное число
if not is_prime(q):
continue
p = 2 * q + 1
if not is_prime(p):
continue
# Найдём генератор
g = ParamGenerator._find_generator(p, q)
return p, q, g
@staticmethod
def _find_generator(p: int, q: int) -> int:
"""
Ищет генератор g: минимальное число 2..p-2,
для которого g^q mod p != 1.
"""
for g in range(2, p - 1):
if pow(g, q, p) != 1:
return g
raise ValueError("Не удалось найти генератор для p")
@@ -0,0 +1,16 @@
from core.participant import Participant
from core.public_params import PublicParams
class Alice:
def __init__(self, params: PublicParams, secret: int | None = None):
self.party = Participant(params, secret=secret)
self.shared_key = None
@property
def A(self) -> int:
return self.party.public_value
def receive_B(self, B: int):
self.shared_key = self.party.compute_shared_key(B)
return self.shared_key
@@ -0,0 +1,16 @@
from core.participant import Participant
from core.public_params import PublicParams
class Bob:
def __init__(self, params: PublicParams, secret: int | None = None):
self.party = Participant(params, secret=secret)
self.shared_key = None
@property
def B(self) -> int:
return self.party.public_value
def receive_A(self, A: int):
self.shared_key = self.party.compute_shared_key(A)
return self.shared_key
+30
View File
@@ -0,0 +1,30 @@
from utils.check_case import check_case
from utils.format_line import format_line
CASES = [
(29, 2, 60, 82, 23),
(53, 2, 20, 77, 42),
(73, 5, 37, 47, 31),
(31, 3, 31, 51, 15),
(41, 6, 31, 98, 8),
(19, 2, 20, 73, 4),
(71, 7, 30, 89, 45),
(109, 6, 7, 14, 84),
(131, 2, 88, 15, 52),
(97, 5, 58, 68, 6),
]
def main():
all_ok = True
for i, case in enumerate(CASES, start=1):
ok, (A, B, K_a, K_b) = check_case(*case)
print(format_line(i, A, B, K_a, K_b, ok))
if not ok:
all_ok = False
if not all_ok:
raise SystemExit(1)
if __name__ == "__main__":
main()
@@ -0,0 +1 @@
#
@@ -0,0 +1,21 @@
from core.public_params import PublicParams
from core.participant import Participant
def check_case(prime_modulus, generator, x, y, K_expected):
"""
Строит участников, имитирует обмен и проверяет совпадение ключа с эталоном.
Возвращает: ok (bool), (A, B, K_a, K_b)
"""
params = PublicParams(prime_modulus, generator)
alice = Participant(params, secret=x)
bob = Participant(params, secret=y)
A = alice.public_value
B = bob.public_value
K_a = alice.compute_shared_key(B)
K_b = bob.compute_shared_key(A)
ok = (K_a == K_b == K_expected)
return ok, (A, B, K_a, K_b)
@@ -0,0 +1,13 @@
def format_line(i, A, B, K_a, K_b, ok):
"""
Форматирует вывод строки для теста.
"""
status = "OK" if ok else "FAIL"
return (
f"#{i:>2}: "
f"A={A:>6} "
f"B={B:>6} "
f"K_A={K_a:>6} "
f"K_B={K_b:>6} "
f"-> {status}"
)
Binary file not shown.
+80
View File
@@ -0,0 +1,80 @@
import json
import sys
from cli.parse_args import parse_args
def main() -> int:
args = parse_args()
try:
if args.cmd == "keygen":
# Класс для режима генерации ключей
from modes.keygen import KeygenMode
runner = KeygenMode()
result = runner.run(
p=args.p,
q=args.q,
min=args.min,
max=args.max,
bits=args.bits,
seed=args.seed,
)
elif args.cmd == "encrypt":
# Класс для режима шифрования
from modes.encrypt import EncryptMode
runner = EncryptMode()
result = runner.run(
text=args.text,
m=args.m,
n=args.n,
e=args.e,
)
elif args.cmd == "decrypt":
# Класс для режима расшифрования
from modes.decrypt import DecryptMode
runner = DecryptMode()
result = runner.run(
c=args.c,
n=args.n,
d=args.d,
)
elif args.cmd == "auto":
from modes.auto import AutoMode
runner = AutoMode()
result = runner.run(
text=args.text,
m=args.m,
p=args.p,
q=args.q,
min=args.min,
max=args.max,
bits=args.bits,
seed=args.seed,
)
else:
raise RuntimeError(f"Неизвестная команда: {args.cmd}")
# Печать результата, если он что-то возвращает
if result is not None:
if isinstance(result, (dict, list)):
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
print(result)
return 0
except Exception as exc:
print(f"Ошибка: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
main()
@@ -0,0 +1,104 @@
from argparse import ArgumentParser, RawDescriptionHelpFormatter
def _int_auto(x: str) -> int:
return int(x, 0)
def parse_args(argv=None):
ap = ArgumentParser(
description=(
"RSA Algorithm CLI with refactored argument handling.\n\n"
"Examples:\n"
" app.py keygen --bits 8\n"
" app.py encrypt --text \"RSA\" --e 17 --n 0xca1\n"
" app.py decrypt --c 0x357 --n 0xca1 --d 0xac1\n"
" app.py auto --text \"RSA\" --bits 8\n"
" app.py auto --m 65 --min 100 --max 1000\n"
),
formatter_class=RawDescriptionHelpFormatter,
)
sub = ap.add_subparsers(dest="cmd", required=True)
# -------- parents to deduplicate args --------
# Общие параметры генерации ключей (используются в keygen и auto)
parent_keygen = ArgumentParser(add_help=False)
parent_keygen.add_argument("--p", type=_int_auto, help="Простое число p.")
parent_keygen.add_argument("--q", type=_int_auto, help="Простое число q.")
parent_keygen.add_argument(
"--min",
type=int,
default=2,
help="Нижняя граница поиска простых (при случайной генерации диапазоном).",
)
parent_keygen.add_argument(
"--max",
type=int,
help="Верхняя граница поиска простых (при случайной генерации диапазоном).",
)
parent_keygen.add_argument(
"--bits",
type=int,
help="Количество бит для случайной генерации простых (альтернатива min/max).",
)
parent_keygen.add_argument(
"--seed", type=int, help="Seed для ГСЧ (опционально, для воспроизводимости)."
)
# Общий модуль n (используется в encrypt и decrypt)
parent_n = ArgumentParser(add_help=False)
parent_n.add_argument("--n", type=_int_auto,
required=True, help="Модуль n.")
# Общий ввод открытого сообщения (используется в encrypt и auto)
parent_plain = ArgumentParser(add_help=False)
plain_group = parent_plain.add_mutually_exclusive_group(required=True)
plain_group.add_argument(
"--text", type=str, help="Открытый текст (ASCII).")
plain_group.add_argument(
"--m", type=_int_auto, help="Открытое сообщение как неотрицательное целое."
)
# -------- keygen --------
sub.add_parser(
"keygen",
help="Сгенерировать приватный (p, q, d, n) и публичный (e, n) ключи.",
formatter_class=RawDescriptionHelpFormatter,
parents=[parent_keygen],
)
# -------- encrypt --------
sp_enc = sub.add_parser(
"encrypt",
help="Зашифровать сообщение публичным ключом (e, n).",
formatter_class=RawDescriptionHelpFormatter,
parents=[parent_n, parent_plain],
)
sp_enc.add_argument(
"--e", type=_int_auto, required=True, help="Публичная экспонента e (простое)."
)
# -------- decrypt --------
sp_dec = sub.add_parser(
"decrypt",
help="Расшифровать шифртекст приватным ключом (n, d).",
formatter_class=RawDescriptionHelpFormatter,
parents=[parent_n],
)
sp_dec.add_argument(
"--c", type=_int_auto, required=True, help="Шифртекст как неотрицательное целое."
)
sp_dec.add_argument(
"--d", type=_int_auto, required=True, help="Приватная экспонента d."
)
# -------- auto --------
sub.add_parser(
"auto",
help="Автоматический режим: как keygen + обязательный ввод сообщения.",
formatter_class=RawDescriptionHelpFormatter,
parents=[parent_keygen, parent_plain],
)
return ap.parse_args(argv)
@@ -0,0 +1,88 @@
import random
import math
class PrimeNumber:
"""
Класс-обёртка для простого числа.
Контракт: внутри всегда хранится валидное простое число.
"""
MAX_ATTEMPTS: int = 1000 # лимит попыток генерации
def __init__(self, value: int):
if not self._is_prime(value):
raise ValueError(f"Число {value} не является простым")
self._value = value
@classmethod
def from_value(cls, value: int) -> "PrimeNumber":
"""Фабричный метод: создать простое число из заданного значения."""
return cls(value)
@classmethod
def random(
cls,
min_value: int | None = None,
max_value: int | None = None,
seed: int | None = None
) -> "PrimeNumber":
"""
Случайное простое число из диапазона [min_value, max_value].
Если seed указан, ГСЧ инициализируется для воспроизводимости.
"""
if min_value < 2:
raise ValueError("Нижняя граница min_value должна быть ≥ 2")
if min_value > max_value:
raise ValueError("min_value не может быть больше max_value")
if seed is not None:
random.seed(seed)
for _ in range(cls.MAX_ATTEMPTS):
candidate = random.randint(min_value, max_value)
if cls._is_prime(candidate):
return cls(candidate)
raise ValueError(
f"Не удалось найти простое число в диапазоне [{
min_value}, {max_value}] "
f"за {cls.MAX_ATTEMPTS} попыток"
)
@classmethod
def random_bits(cls, bits: int, seed: int | None = None) -> "PrimeNumber":
if bits < 2:
raise ValueError("Количество бит должно быть не меньше 2")
# Нижняя граница — 2^(bits-1), верхняя — 2^bits - 1
min_value = 1 << (bits - 1)
max_value = (1 << bits) - 1
return cls.random(
min_value=min_value,
max_value=max_value,
seed=seed,
)
@property
def value(self) -> int:
return self._value
def __repr__(self) -> str:
return f"PrimeNumber({self._value})"
@staticmethod
def _is_prime(n: int) -> bool:
"""Проверка на простоту (НУ ОООЧЕНЬ МЕДЛЕННАЯ)."""
if n < 2:
return False
if n in (2, 3):
return True
if n % 2 == 0:
return False
limit = int(math.isqrt(n)) + 1
for i in range(3, limit, 2):
if n % i == 0:
return False
return True
@@ -0,0 +1,51 @@
from math import gcd
from .prime_number import PrimeNumber
from .public_key import PublicKey
class PrivateKey:
"""
Закрытый ключ RSA.
p1, p2 — экземпляры PrimeNumber.
e — экземпляр PrimeNumber (открытая степень).
n = p1 * p2 (int).
"""
def __init__(self, p1: PrimeNumber, p2: PrimeNumber):
if not isinstance(p1, PrimeNumber) or not isinstance(p2, PrimeNumber):
raise TypeError("p1 и p2 должны быть экземплярами PrimeNumber")
if p1.value == p2.value:
raise ValueError("p1 и p2 должны быть разными простыми")
self.p1 = p1
self.p2 = p2
self.n = p1.value * p2.value
self.phi = (p1.value - 1) * (p2.value - 1)
self.e = self._choose_e()
self.d = self._compute_d()
def _choose_e(self) -> PrimeNumber:
# ищем простое e, взаимнопростое с φ(n)
candidate = 3
while candidate < self.phi:
if gcd(candidate, self.phi) == 1 and PrimeNumber._is_prime(candidate):
return PrimeNumber(candidate)
candidate += 2
raise ValueError("Не удалось подобрать e")
def _compute_d(self) -> int:
k = 1
while True:
if (k * self.phi + 1) % self.e.value == 0:
return (k * self.phi + 1) // self.e.value
k += 1
def make_public_key(self) -> PublicKey:
return PublicKey(self.e, self.n)
def decrypt(self, c: int) -> int:
if not isinstance(c, int) or c < 0:
raise ValueError(
"Шифртекст c должен быть неотрицательным целым числом")
return pow(c, self.d, self.n)
@@ -0,0 +1,23 @@
from .prime_number import PrimeNumber
class PublicKey:
"""
Публичный ключ (e, n).
e — экземпляр PrimeNumber (простое число).
n — целое (составное число p1*p2).
"""
def __init__(self, e: PrimeNumber, n: int):
if not isinstance(e, PrimeNumber):
raise TypeError("e должно быть экземпляром PrimeNumber")
if not isinstance(n, int):
raise TypeError("n должно быть int (составное число)")
self.e = e
self.n = n
def encrypt(self, m: int) -> int:
if not isinstance(m, int) or m < 0:
raise ValueError(
"Сообщение m должно быть неотрицательным целым числом")
return pow(m, self.e.value, self.n)
+68
View File
@@ -0,0 +1,68 @@
from __future__ import annotations
from typing import Dict, Any
from modes.keygen import KeygenMode
from modes.encrypt import EncryptMode
from modes.decrypt import DecryptMode
class AutoMode:
"""
Автоматический режим:
1. Генерация ключей (p, q, n, e, d, phi).
2. Шифрование текста или числа.
3. Расшифрование результата обратно.
"""
def run(
self,
*,
text: str | None,
m: int | None,
p: int | None = None,
q: int | None = None,
min: int | None = None,
max: int | None = None,
bits: int | None = None,
seed: int | None = None,
) -> Dict[str, Any]:
# 1. Генерация ключей
keygen = KeygenMode()
keys = keygen.run(
p=p,
q=q,
min=min,
max=max,
bits=bits,
seed=seed,
)
n = keys["n"]["dec"]
e = keys["e"]["dec"]
d = keys["d"]["dec"]
# 2. Шифрование
encrypt = EncryptMode()
encrypted = encrypt.run(
text=text,
m=m,
n=n,
e=e,
)
# 3. Расшифрование
decrypt = DecryptMode()
# В зависимости от формата encrypted["output"]
if encrypted["mode"] == "text":
ciphers = [item["cipher"] for item in encrypted["output"]]
decrypted = decrypt.run(c=ciphers, n=n, d=d)
else:
cipher = encrypted["output"]["cipher"]
decrypted = decrypt.run(c=cipher, n=n, d=d)
return {
"keys": keys,
"encrypted": encrypted,
"decrypted": decrypted,
}
@@ -0,0 +1,55 @@
# modes/decrypt.py
from __future__ import annotations
from typing import Dict, Any, List
class DecryptMode:
def run(
self,
*,
c: int | List[int],
n: int,
d: int,
) -> Dict[str, Any]:
# Один блок
if isinstance(c, int):
plain = pow(c, d, n) # Дешифратор
char = chr(plain) if 0 <= plain < 128 else None
return {
"mode": "int",
"input": {
"cipher": c,
"hex_in": hex(c),
},
"output": {
"ascii": plain,
"char": char,
"hex_out": hex(plain),
},
}
# Несколько блоков (список чисел)
if isinstance(c, list):
result: List[Dict[str, Any]] = []
text = ""
for block in c:
plain = pow(block, d, n) # Дешифратор
char = chr(plain) if 0 <= plain < 128 else None
if char:
text += char
result.append({
"cipher": block,
"hex_in": hex(block),
"ascii": plain,
"char": char,
"hex_out": hex(plain),
})
return {
"mode": "list",
"input": c,
"output": result,
"text": text,
}
raise ValueError("c должен быть числом или списком чисел")
@@ -0,0 +1,57 @@
from __future__ import annotations
from typing import Dict, Any, List
from core.public_key import PublicKey
from core.prime_number import PrimeNumber
class EncryptMode:
def run(
self,
*,
text: str | None,
m: int | None,
n: int,
e: int,
) -> Dict[str, Any]:
pub = PublicKey(n=n, e=PrimeNumber(e))
if text is not None:
result: List[Dict[str, Any]] = []
for ch in text:
code = ord(ch)
cipher = pub.encrypt(code)
result.append({
"char": ch,
"ascii": code,
"hex_in": hex(code),
"cipher": cipher,
"hex_out": hex(cipher),
})
return {
"mode": "text",
"input": text,
"output": result,
}
if m is not None:
cipher = pub.encrypt(m)
char = chr(m) if 0 <= m < 128 else None
return {
"mode": "int",
"input": {
"char": char,
"ascii": m,
"hex_in": hex(m),
},
"output": {
"char": char,
"ascii": m,
"cipher": cipher,
"hex_out": hex(cipher),
},
}
raise ValueError("Необходимо указать либо text, либо m")
+133
View File
@@ -0,0 +1,133 @@
from __future__ import annotations
from typing import Dict, Any
from core.prime_number import PrimeNumber
from core.private_key import PrivateKey
class KeygenMode:
MAX_BITS: int = 2**5
MIN_VALUE: int = 2
MAX_VALUE: int = (1 << MAX_BITS) - 1
MAX_ATTEMPTS: int = 1000
def run(
self,
*,
p: int | None,
q: int | None,
min: int | None = None,
max: int | None = None,
bits: int | None = None,
seed: int | None = None,
) -> Dict[str, Any]:
if min is None:
min = self.MIN_VALUE
if max is None:
max = self.MAX_VALUE
self._validate_inputs(p, q, min, max, bits)
P = self._generate_prime(p, min, max, bits, seed)
Q = self._generate_distinct_prime(P, q, min, max, bits, seed)
priv = PrivateKey(P, Q)
pub = priv.make_public_key()
return {
"p": {
"dec": P.value,
"hex": hex(P.value),
},
"q": {
"dec": Q.value,
"hex": hex(Q.value),
},
"n": {
"dec": priv.n,
"hex": hex(priv.n),
},
"phi": {
"dec": priv.phi,
"hex": hex(priv.phi),
},
"e": {
"dec": pub.e.value,
"hex": hex(pub.e.value),
},
"d": {
"dec": priv.d,
"hex": hex(priv.d),
},
}
# ----------------- helpers -----------------
def _validate_inputs(
self,
p: int | None,
q: int | None,
min: int,
max: int,
bits: int | None,
) -> None:
if p is not None and q is not None and p == q:
raise ValueError("p и q должны быть разными простыми числами.")
if bits is not None and bits > self.MAX_BITS:
raise ValueError(f"Максимальная длина простого: {
self.MAX_BITS} бит.")
if min > max:
raise ValueError("min не может быть больше max.")
if max > self.MAX_VALUE:
raise ValueError(
f"max не может превышать {self.MAX_VALUE} (ограничение {
self.MAX_BITS} бит)."
)
if min == max:
raise ValueError("min и max не должны совпадать.")
def _generate_prime(
self,
value: int | None,
min: int,
max: int,
bits: int | None,
seed: int | None,
) -> PrimeNumber:
if value is not None:
return PrimeNumber.from_value(value)
if bits is not None:
return PrimeNumber.random_bits(bits=bits, seed=seed)
else:
return PrimeNumber.random(min_value=min, max_value=max, seed=seed)
return PrimeNumber.random_bits(bits=self.MAX_BITS, seed=seed)
def _generate_distinct_prime(
self,
other: PrimeNumber,
value: int | None,
min: int,
max: int,
bits: int | None,
seed: int | None,
) -> PrimeNumber:
attempts = 0
while True:
effective_seed = None if seed is None else seed + attempts
candidate = self._generate_prime(
value, min, max, bits, effective_seed)
if candidate.value != other.value:
return candidate
attempts += 1
if attempts >= self.MAX_ATTEMPTS:
raise RuntimeError(
f"Не удалось сгенерировать q, отличное от p, за {
self.MAX_ATTEMPTS} попыток."
)
@@ -0,0 +1,26 @@
import pytest
from core.prime_number import PrimeNumber
def test_from_value_valid():
p = PrimeNumber.from_value(61)
assert p.value == 61
assert repr(p) == "PrimeNumber(61)"
def test_from_value_invalid():
with pytest.raises(ValueError):
PrimeNumber.from_value(60) # не простое
def test_random_in_range():
p = PrimeNumber.random(2, 20)
assert isinstance(p, PrimeNumber)
assert 2 <= p.value <= 20
# дополнительно проверим, что оно действительно простое
assert PrimeNumber._is_prime(p.value)
def test_random_no_primes():
with pytest.raises(ValueError):
PrimeNumber.random(90, 92) # в этом диапазоне простых нет
@@ -0,0 +1,50 @@
import pytest
from core.private_key import PrivateKey
from core.prime_number import PrimeNumber
def test_encrypt_decrypt_cycle():
"""Проверка: сообщение после encrypt → decrypt восстанавливается"""
p1 = PrimeNumber.from_value(61)
p2 = PrimeNumber.from_value(53)
priv = PrivateKey(p1, p2)
pub = priv.make_public_key()
messages = [1, 2, 42, 123, priv.n - 1]
for m in messages:
c = pub.encrypt(m)
m_back = priv.decrypt(c)
assert m == m_back
def test_e_is_prime_and_coprime_with_phi():
"""e должно быть простым и взаимно простым с φ(n)"""
p1 = PrimeNumber.from_value(61)
p2 = PrimeNumber.from_value(53)
priv = PrivateKey(p1, p2)
assert isinstance(priv.e, PrimeNumber)
assert (priv.e.value > 1)
assert (priv.phi % priv.e.value != 0)
def test_d_is_inverse_of_e_mod_phi():
"""e*d ≡ 1 (mod φ(n))"""
p1 = PrimeNumber.from_value(61)
p2 = PrimeNumber.from_value(53)
priv = PrivateKey(p1, p2)
assert (priv.e.value * priv.d) % priv.phi == 1
def test_decrypt_invalid_inputs():
"""Должны бросаться ошибки при неверных входных данных"""
p1 = PrimeNumber.from_value(61)
p2 = PrimeNumber.from_value(53)
priv = PrivateKey(p1, p2)
with pytest.raises(ValueError):
priv.decrypt(-5)
with pytest.raises(ValueError):
priv.decrypt("not a number") # type: ignore
@@ -0,0 +1,47 @@
import pytest
from core.prime_number import PrimeNumber
from core.public_key import PublicKey
def test_encrypt_basic_cycle():
"""Шифрование совпадает с pow(m, e, n)"""
e = PrimeNumber.from_value(3)
n = 11 # n может быть простым или составным числом
pub = PublicKey(e, n)
for m in range(1, 10):
expected = pow(m, e.value, n)
assert pub.encrypt(m) == expected
def test_encrypt_with_composite_n():
"""n = p1 * p2 должно работать корректно"""
p1 = PrimeNumber.from_value(5)
p2 = PrimeNumber.from_value(11)
n = p1.value * p2.value
e = PrimeNumber.from_value(3)
pub = PublicKey(e, n)
m = 7
expected = pow(m, e.value, n)
assert pub.encrypt(m) == expected
def test_encrypt_rejects_negative():
"""Отрицательные сообщения запрещены"""
e = PrimeNumber.from_value(3)
n = 11
pub = PublicKey(e, n)
with pytest.raises(ValueError):
pub.encrypt(-1)
def test_encrypt_rejects_non_integer():
"""Сообщение должно быть int"""
e = PrimeNumber.from_value(3)
n = 11
pub = PublicKey(e, n)
with pytest.raises(ValueError):
pub.encrypt(3.14)
Binary file not shown.
+68
View File
@@ -0,0 +1,68 @@
import json
import sys
from cli.parse_args import parse_args
def main() -> int:
args = parse_args()
try:
if args.cmd == "hash":
from modes.hash_mode import HashMode
runner = HashMode()
result = runner.run(
variant=args.variant,
message=args.message,
salt=args.salt,
bytes=args.bytes,
)
elif args.cmd == "avalanche":
from modes.test_avalanche import AvalancheTest
runner = AvalancheTest()
result = runner.run(
variant=args.variant,
message=args.message,
salt=args.salt,
bytes=args.bytes,
)
elif args.cmd == "collisions":
from modes.test_collisions import CollisionsTest
runner = CollisionsTest()
result = runner.run(
variant=args.variant,
max_len=args.max_len,
alphabet=args.alphabet,
bytes=args.bytes,
)
elif args.cmd == "analyze":
from modes.analyze_uniformity import UniformityAnalyzer
runner = UniformityAnalyzer()
result = runner.run(
variant=args.variant,
count=args.count,
buckets=args.buckets,
)
else:
raise RuntimeError(f"Неизвестная команда: {args.cmd}")
# Печать результата
if result is not None:
if isinstance(result, (dict, list)):
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
print(result)
return 0
except Exception as exc:
print(f"Ошибка: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
main()
+53
View File
@@ -0,0 +1,53 @@
from argparse import ArgumentParser
def parse_args(argv=None):
ap = ArgumentParser(
prog="app.py",
description="CLI для лабораторной по хэшированию"
)
sub = ap.add_subparsers(dest="cmd", required=True)
# --- Общие группы ---
common_variant = ArgumentParser(add_help=False)
common_variant.add_argument("-v", "--variant", choices=["1", "2", "3"], required=True,
help="Выбор варианта: 1=sin, 2=cos, 3=xor")
common_salt = ArgumentParser(add_help=False)
common_salt.add_argument("-s", "--salt", type=str, default="",
help="Соль (по умолчанию пустая)")
common_bytes = ArgumentParser(add_help=False)
common_bytes.add_argument("--bytes", type=int, default=16,
help="Количество значимых байтов хэша (по умолчанию 16)")
# --- HASH ---
sp_hash = sub.add_parser("hash", parents=[common_variant, common_salt, common_bytes],
help="Посчитать хэш сообщения")
sp_hash.add_argument("-m", "--message", type=str, default="",
help="Сообщение (может быть пустым)")
# --- AVALANCHE ---
sp_av = sub.add_parser("avalanche", parents=[common_variant, common_salt, common_bytes],
help="Тест эффекта лавины")
sp_av.add_argument("-m", "--message", type=str, required=True,
help="Базовое сообщение")
# --- COLLISIONS ---
sp_col = sub.add_parser("collisions", parents=[common_variant, common_bytes],
help="Поиск коллизий (перебор)")
sp_col.add_argument("--max-len", type=int, default=3,
help="Максимальная длина сообщений (по умолчанию 3)")
sp_col.add_argument("--alphabet", choices=["alnum", "alpha", "digits"], default="alnum",
help="Алфавит: alnum[a-z0-9], alpha[a-z], digits[0-9]")
# --- ANALYZE ---
sp_an = sub.add_parser("analyze", parents=[common_variant, common_bytes],
help="Анализ равномерности")
sp_an.add_argument("--count", type=int, default=100,
help="Сколько сообщений сгенерировать")
sp_an.add_argument("--buckets", type=int, default=10,
help="Сколько корзин для распределения (по умолчанию 10)")
return ap.parse_args(argv)
@@ -0,0 +1,44 @@
import string
import random
from modes.hash_mode import HashMode
class UniformityAnalyzer:
"""
Анализ равномерности распределения хэшей.
"""
def run(self, variant: str, count: int = 100, buckets: int = 10, bytes: int = 16) -> dict:
hasher = HashMode()
alphabet = string.ascii_lowercase + string.digits
# случайные сообщения
messages = [
"".join(random.choice(alphabet)
for _ in range(random.randint(5, 10)))
for _ in range(count)
]
# хэши
hashes = [hasher.run(variant, msg, bytes=bytes) for msg in messages]
# берём первые 8 hex-символов → число
values = [int(h[:8], 16) for h in hashes]
values.sort()
# диапазоны
min_val, max_val = values[0], values[-1]
step = (max_val - min_val) / buckets if max_val > min_val else 1
distribution = [0] * buckets
for v in values:
idx = min(int((v - min_val) / step), buckets - 1)
distribution[idx] += 1
return {
"count": count,
"buckets": buckets,
"messages": messages,
"hashes": hashes,
"distribution": distribution,
}
+95
View File
@@ -0,0 +1,95 @@
# modes/hash_mode.py
import math
class HashMode:
"""
Учебная хэш-функция.
Варианты:
1 = sin-based
2 = cos-based
3 = xor-based
"""
def run(self, variant: str, message: str, salt: str = "", bytes: int = 16) -> str:
if not (2 <= bytes <= 16):
raise ValueError("Параметр bytes должен быть в диапазоне [2, 16]")
data = (message + salt).encode("utf-8")
if variant == "1":
out = self._sin_hash(data)
elif variant == "2":
out = self._cos_hash(data)
elif variant == "3":
out = self._xor_hash(data)
else:
raise ValueError(f"Неизвестный вариант: {variant}")
# Обрезаем до нужного числа байт
truncated = self._truncate_bytes(out, bytes)
return ''.join(f"{x:02x}" for x in truncated)
# --- служебные ------------------------------------------------------
def _truncate_bytes(self, out: bytearray, bytes: int) -> bytearray:
if bytes >= len(out):
return out
truncated = out[:bytes]
# примешиваем все отброшенные байты в оставшиеся
for i, b in enumerate(out[bytes:], start=1):
truncated[i % len(truncated)] ^= b
return truncated
# --- реализации ------------------------------------------------------
def _sin_hash(self, data: bytes) -> bytearray:
out = [0x55 ^ i for i in range(16)]
if not data:
return out
angle = (2 * math.pi) / (len(out) * len(data))
for i, b in enumerate(data, 1):
for j in range(len(out)):
val = int((math.sin(b + (i + j) * angle) + 1) * 127)
out[j] ^= val & 0xFF
return out
def _cos_hash(self, data: bytes) -> bytearray:
out = [0x77 ^ i for i in range(16)]
if not data:
return out
angle = (2 * math.pi) / (len(out) * len(data))
for i, b in enumerate(data, 1):
for j in range(len(out)):
val = int((math.cos(b + (i + j) * angle) + 1) * 127)
out[j] ^= val & 0xFF
return out
def _mix_bits(self, mix: int) -> bytearray:
# количество байт у числа = (битовая длина + 7) // 8
count = (mix.bit_length() + 7) // 8
for i in range(1, count):
mix ^= (mix >> (8 * i))
return mix
def _xor_hash(self, data: bytes) -> bytearray:
out = [0xAA ^ i for i in range(16)]
if not data:
return out
phi = (1 + 5 ** 0.5) / 2 # золотое сечение
for i, b in enumerate(data, 1):
for j in range(16):
mix = (b * (i + j + 1)) ^ int(phi * 1e6)
mix = self._mix_bits(mix)
out[j] ^= mix & 0xFF # только младший байт
return out
@@ -0,0 +1,39 @@
from modes.hash_mode import HashMode
class AvalancheTest:
"""
Проверка эффекта лавины:
небольшое изменение во входном сообщении должно сильно менять хэш.
"""
def run(self, variant: str, message: str, salt: str = "", bytes: int = 16) -> dict:
hasher = HashMode()
# базовый хэш
h1 = hasher.run(variant, message, salt=salt, bytes=bytes)
# изменяем один символ (если пусто — подставим "a")
if message:
i = 0 # всегда первый символ для простоты
flipped_char = chr((ord(message[i]) + 1) % 128)
mod_msg = flipped_char + message[1:]
else:
mod_msg = "a"
h2 = hasher.run(variant, mod_msg, salt=salt, bytes=bytes)
# сравнение побитово
diff_bits = sum(c1 != c2 for c1, c2 in zip(
bin(int(h1, 16))[2:], bin(int(h2, 16))[2:]))
total_bits = len(h1) * 4 # HEX -> биты
return {
"original": message,
"modified": mod_msg,
"hash1": h1,
"hash2": h2,
"diff_bits": diff_bits,
"total_bits": total_bits,
"diff_percent": round(100 * diff_bits / total_bits, 2),
}
@@ -0,0 +1,38 @@
import itertools
import string
from modes.hash_mode import HashMode
class CollisionsTest:
"""
Грубый поиск коллизий: перебираем строки до max_len
из заданного алфавита и проверяем, совпадут ли хэши.
"""
def run(self, variant: str, max_len: int = 3, alphabet: str = "alnum", bytes: int = 16) -> dict:
hasher = HashMode()
seen = {}
# выбор алфавита
if alphabet == "alnum":
chars = string.ascii_lowercase + string.digits
elif alphabet == "alpha":
chars = string.ascii_lowercase
elif alphabet == "digits":
chars = string.digits
else:
raise ValueError(f"Неизвестный алфавит: {alphabet}")
for length in range(1, max_len + 1):
for msg_tuple in itertools.product(chars, repeat=length):
msg = "".join(msg_tuple)
h = hasher.run(variant, msg, bytes=bytes)
if h in seen:
return {
"collision_found": True,
"hash": h,
"messages": [seen[h], msg],
}
seen[h] = msg
return {"collision_found": False}
Binary file not shown.
@@ -0,0 +1,89 @@
import subprocess
import json
import sys
from pathlib import Path
from cli.parse_args import parse_args
LAB2 = Path(__file__).resolve().parents[2] / \
"lab2-the-RSA-algorithm" / "app" / "app.py"
LAB3 = Path(__file__).resolve().parents[2] / \
"lab3-hash-function" / "app" / "app.py"
def run_lab2_keygen(bits: int, seed: int | None):
cmd = [sys.executable, str(LAB2), "keygen", "--bits", str(bits)]
if seed is not None:
cmd += ["--seed", str(seed)]
res = subprocess.run(cmd, capture_output=True, text=True, check=True)
return json.loads(res.stdout)
def run_lab3_hash(variant: str, text: str, bytes_len: int):
cmd = [
sys.executable,
str(LAB3),
"hash",
"-v",
variant,
"-m",
text,
"--bytes",
str(bytes_len),
]
res = subprocess.run(cmd, capture_output=True, text=True, check=True)
return res.stdout.strip() # ожидается hex-строка
def main():
args = parse_args()
# Фиксированные параметры RSA и HASH
bits = 32
bytes_len = 7
# 1. Генерация ключей RSA
keys = run_lab2_keygen(bits, args.seed)
n = int(keys["n"]["dec"])
e = int(keys["e"]["dec"])
d = int(keys["d"]["dec"])
print("RSA ключи сгенерированы:")
print(json.dumps(keys, indent=2, ensure_ascii=False))
# 2. Хэш исходного текста
h_hex = run_lab3_hash(args.variant, args.text, bytes_len)
h = int(h_hex, 16)
print(f"\nХэш исходного сообщения: {h_hex} (dec={h})")
# 3. Проверка условия n > h
if h >= n:
print(
"\n[ОШИБКА] Хэш больше или равен n. Нужно пересоздать ключи с большим размером.")
sys.exit(1)
# 4. Подпись: S = h^d mod n
S = pow(h, d, n)
print(f"\nПодпись (S): {S}")
# 5. Проверка подписи для исходного сообщения
h_check = pow(S, e, n)
print(f"Проверка подписи: {h_check} == {h} ? {
'OK' if h_check == h else 'FAIL'}")
# 6. Если передан модифицированный текст — проверить и его
if args.tampered:
h_tampered_hex = run_lab3_hash(args.variant, args.tampered, bytes_len)
h_tampered = int(h_tampered_hex, 16)
print(f"\nХэш модифицированного сообщения: {
h_tampered_hex} (dec={h_tampered})")
h_check_tampered = pow(S, e, n)
print(
f"Проверка подписи для модифицированного текста: "
f"{h_check_tampered} == {h_tampered} ? {
'OK' if h_check_tampered == h_tampered else 'FAIL'}"
)
if __name__ == "__main__":
main()
@@ -0,0 +1,25 @@
from argparse import ArgumentParser
def parse_args(argv=None):
ap = ArgumentParser(
prog="app.py",
description="ЭЦП (RSA + собственная хэш-функция). "
"Выполняет полный цикл: генерация ключей, подпись и проверку."
)
# --- RSA ---
ap.add_argument("-s", "--seed", type=int,
help="Seed для ГСЧ (опционально, для воспроизводимости).")
# --- HASH ---
ap.add_argument("-v", "--variant", choices=["1", "2", "3"], required=True,
help="Вариант хэш-функции (см. lab3).")
# --- MESSAGES ---
ap.add_argument("-t", "--text", type=str, required=True,
help="Исходное сообщение (ASCII).")
ap.add_argument("-m", "--tampered", type=str,
help="Модифицированное сообщение для имитации нарушения целостности.")
return ap.parse_args(argv)
+34
View File
@@ -0,0 +1,34 @@
from core import PasswordHasher, IntegrityChecker, CryptoManager, AuditLogger, AccountManager
from gui import MainWindow
class App:
"""
Класс точки входа для запуска менеджера учётных записей.
"""
def __init__(self):
self.db_file = None
self.hash_file = "data/accounts.db.hash"
self.log_file = "data/audit.log"
self.password_hasher = PasswordHasher()
self.integrity_checker = IntegrityChecker()
self.crypto_manager = CryptoManager()
self.audit_logger = AuditLogger(self.log_file)
self.account_manager = AccountManager(self.db_file)
self.gui = MainWindow(
account_manager=self.account_manager,
password_hasher=self.password_hasher,
audit_logger=self.audit_logger,
crypto_manager=self.crypto_manager,
integrity_checker=self.integrity_checker,
)
def run(self):
self.gui.run()
if __name__ == "__main__":
App().run()
@@ -0,0 +1,5 @@
from .password_hasher import PasswordHasher
from .integrity_checker import IntegrityChecker
from .crypto_manager import CryptoManager
from .audit_logger import AuditLogger
from .account_manager import AccountManager
@@ -0,0 +1,62 @@
import json
from pathlib import Path
class AccountManager:
"""
Класс для управления учётными записями пользователей.
"""
def __init__(self, db_file: str | None):
self.db_file = Path(db_file) if db_file else None
if isinstance(self.db_file, Path) and not self.db_file.exists():
self._save_db({})
def set_db_file(self, db_file: str | None) -> None:
self.db_file = Path(db_file) if db_file else None
if isinstance(self.db_file, Path) and not self.db_file.exists():
self._save_db({})
def _load_db(self) -> dict:
if not isinstance(self.db_file, Path):
return {}
if self.db_file.exists():
return json.loads(self.db_file.read_text(encoding="utf-8"))
return {}
def _save_db(self, data: dict) -> None:
if not isinstance(self.db_file, Path):
raise RuntimeError("База данных не открыта")
self.db_file.write_text(json.dumps(
data, indent=2, ensure_ascii=False), encoding="utf-8")
def add_user(self, username: str, password_hash: str, salt: str) -> None:
data = self._load_db()
data[username] = {"hash": password_hash, "salt": salt}
self._save_db(data)
def update_user(self, username: str, password_hash: str, salt: str) -> bool:
data = self._load_db()
if username not in data:
return False
data[username] = {"hash": password_hash, "salt": salt}
self._save_db(data)
return True
def delete_user(self, username: str) -> bool:
data = self._load_db()
if username not in data:
return False
del data[username]
self._save_db(data)
return True
def get_user(self, username: str) -> dict | None:
data = self._load_db()
return data.get(username)
def list_users(self) -> list[str]:
if not isinstance(self.db_file, Path):
return []
data = self._load_db()
return list(data.keys())
@@ -0,0 +1,17 @@
from datetime import datetime
from pathlib import Path
class AuditLogger:
"""
Класс для ведения журнала событий.
"""
def __init__(self, log_file: str):
self.log_file = Path(log_file)
def log(self, event: str) -> None:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{timestamp}] {event}\n"
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(line)
@@ -0,0 +1,23 @@
from cryptography.fernet import Fernet
from pathlib import Path
class CryptoManager:
"""
Класс для симметричного шифрования и расшифрования файлов.
"""
def generate_key(self) -> bytes:
return Fernet.generate_key()
def encrypt_file(self, file_path: str, key: bytes, output_path: str) -> None:
fernet = Fernet(key)
data = Path(file_path).read_bytes()
encrypted = fernet.encrypt(data)
Path(output_path).write_bytes(encrypted)
def decrypt_file(self, file_path: str, key: bytes, output_path: str) -> None:
fernet = Fernet(key)
data = Path(file_path).read_bytes()
decrypted = fernet.decrypt(data)
Path(output_path).write_bytes(decrypted)
@@ -0,0 +1,26 @@
import hashlib
from pathlib import Path
class IntegrityChecker:
"""
Класс для проверки целостности файлов по хэш-сумме.
"""
def calculate_hash(self, file_path: str) -> str:
hasher = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
return hasher.hexdigest()
def save_hash(self, file_path: str, hash_file: str) -> None:
file_hash = self.calculate_hash(file_path)
Path(hash_file).write_text(file_hash, encoding="utf-8")
def verify_hash(self, file_path: str, hash_file: str) -> bool:
if not Path(hash_file).exists():
return False
expected = Path(hash_file).read_text(encoding="utf-8").strip()
actual = self.calculate_hash(file_path)
return expected == actual
@@ -0,0 +1,19 @@
import hashlib
import secrets
class PasswordHasher:
"""
Класс для безопасного хэширования паролей с использованием соли.
"""
def generate_salt(self, length: int = 16) -> str:
return secrets.token_hex(length)
def hash_password(self, password: str, salt: str) -> str:
hasher = hashlib.sha256()
hasher.update(salt.encode("utf-8") + password.encode("utf-8"))
return hasher.hexdigest()
def verify_password(self, password: str, salt: str, expected_hash: str) -> bool:
return self.hash_password(password, salt) == expected_hash
@@ -0,0 +1,6 @@
{
"user2": {
"hash": "4d3596717c7887c07c607485fa8c287547a5e65559b3269844261417b1e7a97b",
"salt": "a1bb56a86e8d504634668696d9e65189"
}
}
@@ -0,0 +1 @@
109fb5ce85641ad0ed60cf8ae800289517d33c2cb100f39e5c7c74cb88b9b722
@@ -0,0 +1,3 @@
[2025-10-01 21:51:42] Добавлен пользователь: user
[2025-10-01 21:51:48] Добавлен пользователь: user2
[2025-10-01 21:51:55] Добавлен пользователь: user3
@@ -0,0 +1 @@
gAAAAABo3Wp_Jc6ye1Yf6rYq3GQ64_S6XHAQjm7WSCvS6fK48kMk5NmP7zFyX5AGShawV7uvFEVvx2v5Efn_hFWdML4I_iAf1x6jZheZLJuWzMLtsm4x25wEAo7c8vokR8YV-MaiJWKw_c9kdAzPoPpkfuZSAuwSU6hD_oNHooG8SPCWVTbbvmqwnquMchLpXaK_gfyYOgye_ERB5s5GBxMICwM47W87tGoKnFL6jtyGk8iIowPYD0xlrAhUwvuTeDfxlIhyOZBc72R5iTmiSbOgxNu6DyJy9-49kdSau35LTVWRThg7MEWxK1xTC5P6stXkhovNsjujsdcSBk8Rx-AgKBmSRPGznkC7zYTNw6gQhYujsi_WtaQ=
@@ -0,0 +1 @@
ir7gZ6qT1gP-wAcNvKB3Ua_k7zJ52bH_GmgdSO9iryo=
@@ -0,0 +1,3 @@
from .main_window import MainWindow
__all__ = ["MainWindow"]
@@ -0,0 +1,16 @@
import tkinter as tk
class ControlsPanel:
"""
Панель управления действиями над пользователями.
"""
def __init__(self, master, on_add, on_update, on_delete):
self.frame = tk.Frame(master)
tk.Button(self.frame, text="Добавить",
command=on_add).pack(side=tk.LEFT, padx=5)
tk.Button(self.frame, text="Изменить",
command=on_update).pack(side=tk.LEFT, padx=5)
tk.Button(self.frame, text="Удалить", command=on_delete).pack(
side=tk.LEFT, padx=5)
@@ -0,0 +1,92 @@
from __future__ import annotations
from pathlib import Path
from tkinter import messagebox, filedialog as fd
class CryptoTools:
"""Операции шифрования и расшифрования файлов."""
def __init__(self, crypto_manager, audit_logger):
self.crypto_manager = crypto_manager
self.audit_logger = audit_logger
def _generate_and_save_key(self) -> bytes | None:
key = self.crypto_manager.generate_key()
path = fd.asksaveasfilename(
title="Сохранить ключ шифрования",
defaultextension=".key",
filetypes=[("Key file", "*.key"), ("All files", "*.*")],
initialfile="encryption.key",
)
if not path:
return None
try:
Path(path).write_bytes(key)
except Exception as e:
messagebox.showerror("Сохранение ключа",
f"Не удалось сохранить ключ:\n{e}")
return None
messagebox.showinfo("Ключ сохранён", f"Ключ сохранён в:\n{path}")
return key
def _load_key_from_file(self) -> bytes | None:
path = fd.askopenfilename(
title="Выберите файл ключа",
filetypes=[("Key file", "*.key"), ("All files", "*.*")],
)
if not path:
return None
try:
return Path(path).read_bytes()
except Exception as e:
messagebox.showerror(
"Чтение ключа", f"Не удалось прочитать ключ:\n{e}")
return None
def _select_output_path(self, suggested: Path, title: str, defext: str, pattern: str) -> Path | None:
path = fd.asksaveasfilename(
title=title,
defaultextension=defext,
filetypes=[(f"*{defext}", pattern), ("All files", "*.*")],
initialdir=str(suggested.parent),
initialfile=suggested.name,
)
return Path(path) if path else None
def encrypt_file(self):
src = fd.askopenfilename(title="Выберите файл для шифрования")
if not src:
return
key = self._generate_and_save_key()
if key is None:
return
suggested = Path(src).with_suffix(Path(src).suffix + ".enc")
out = self._select_output_path(
suggested, "Сохранить зашифрованный файл", ".enc", "*.enc")
if out is None:
return
self.crypto_manager.encrypt_file(src, key, str(out))
self.audit_logger.log(f"Зашифрован файл → {out.name}")
messagebox.showinfo("Шифрование файла", f"Создан файл: {out}")
def decrypt_file(self):
enc = fd.askopenfilename(title="Выберите зашифрованный файл", filetypes=[
("Encrypted", "*.enc"), ("All files", "*.*")])
if not enc:
return
key = self._load_key_from_file()
if key is None:
return
suggested = Path(enc).with_suffix(".dec")
out = self._select_output_path(
suggested, "Сохранить расшифрованный файл", ".dec", "*.dec")
if out is None:
return
try:
self.crypto_manager.decrypt_file(enc, key, str(out))
except Exception as e:
messagebox.showerror("Расшифрование файла", f"Ошибка:\n{e}")
return
self.audit_logger.log(f"Расшифрован файл → {out.name}")
messagebox.showinfo("Расшифрование файла", f"Создан файл: {out}")
@@ -0,0 +1,73 @@
from __future__ import annotations
from pathlib import Path
from tkinter import messagebox, filedialog as fd
class IntegrityTools:
"""Операции сохранения и проверки хэша БД."""
def __init__(self, integrity_checker, audit_logger, db_path: str | None):
self.integrity_checker = integrity_checker
self.audit_logger = audit_logger
self.db_path = Path(db_path) if db_path else None
def set_db_path(self, db_path: str | None) -> None:
self.db_path = Path(db_path) if db_path else None
def _hash_path(self) -> Path:
return self.db_path.with_suffix(self.db_path.suffix + ".hash")
def save_db_hash(self):
if not self.db_path:
messagebox.showwarning("Сохранить хэш", "База данных не выбрана.")
return
if not self.db_path.exists():
messagebox.showwarning(
"Сохранить хэш", f"Файл БД не найден:\n{self.db_path}")
return
suggested = self.db_path.with_suffix(self.db_path.suffix + ".hash")
path = fd.asksaveasfilename(
title="Сохранить контрольную сумму БД",
defaultextension=".hash",
filetypes=[("Hash file", "*.hash"), ("All files", "*.*")],
initialdir=str(suggested.parent),
initialfile=suggested.name,
)
if not path:
return
try:
self.integrity_checker.save_hash(str(self.db_path), str(path))
except Exception as e:
messagebox.showerror("Сохранить хэш", f"Ошибка сохранения:\n{e}")
return
self.audit_logger.log(
f"Сохранена контрольная сумма БД → {Path(path).name}")
messagebox.showinfo("Сохранить хэш", f"Хэш сохранён в {path}")
def verify_db_hash(self):
if not self.db_path:
messagebox.showwarning(
"Проверка целостности", "База данных не выбрана.")
return
if not self.db_path.exists():
messagebox.showwarning(
"Проверка целостности", f"Файл БД не найден:\n{self.db_path}")
return
suggested = self.db_path.with_suffix(self.db_path.suffix + ".hash")
path = fd.askopenfilename(
title="Выбрать файл контрольной суммы БД",
filetypes=[("Hash file", "*.hash"), ("All files", "*.*")],
initialdir=str(suggested.parent),
initialfile=suggested.name,
)
if not path:
return
ok = self.integrity_checker.verify_hash(str(self.db_path), str(path))
self.audit_logger.log("Проверка целостности БД: " +
("OK" if ok else "НЕСОВПАДЕНИЕ"))
if ok:
messagebox.showinfo("Проверка целостности", "БД в целостности.")
else:
messagebox.showerror("Проверка целостности",
"БД была изменена извне.")
@@ -0,0 +1,135 @@
from __future__ import annotations
import tkinter as tk
from tkinter import messagebox, simpledialog, filedialog as fd
from pathlib import Path
from .user_list_view import UserListView
from .controls_panel import ControlsPanel
from .integrity_tools import IntegrityTools
from .crypto_tools import CryptoTools
class MainWindow:
"""
Главное окно менеджера учётных записей.
"""
def __init__(self, account_manager, password_hasher, audit_logger, crypto_manager, integrity_checker):
self.account_manager = account_manager
self.password_hasher = password_hasher
self.audit_logger = audit_logger
self.integrity_tools = IntegrityTools(
integrity_checker, audit_logger, account_manager.db_file)
self.crypto_tools = CryptoTools(crypto_manager, audit_logger)
self.root = tk.Tk()
self.root.title("Менеджер учётных записей")
self._build_menu()
self.users_view = UserListView(self.root)
self.users_view.frame.pack()
self.controls = ControlsPanel(
self.root,
on_add=self.add_user,
on_update=self.update_user,
on_delete=self.delete_user
)
self.controls.frame.pack(pady=5)
self.refresh_users()
def _build_menu(self):
menubar = tk.Menu(self.root)
file_menu = tk.Menu(menubar, tearoff=0)
file_menu.add_command(
label="Открыть базу данных...", command=self.open_db)
file_menu.add_command(label="Закрыть базу данных",
command=self.close_db)
menubar.add_cascade(label="Файл", menu=file_menu)
tools = tk.Menu(menubar, tearoff=0)
tools.add_command(label="Сохранить хэш БД",
command=self.integrity_tools.save_db_hash)
tools.add_command(label="Проверить целостность БД",
command=self.integrity_tools.verify_db_hash)
tools.add_separator()
tools.add_command(label="Зашифровать файл...",
command=self.crypto_tools.encrypt_file)
tools.add_command(label="Расшифровать файл...",
command=self.crypto_tools.decrypt_file)
menubar.add_cascade(label="Инструменты", menu=tools)
self.root.config(menu=menubar)
# --- File ops ---
def open_db(self):
path = fd.askopenfilename(
title="Открыть базу данных",
filetypes=[("JSON", "*.json"), ("DB/TXT", "*.db *.txt"),
("Все файлы", "*.*")],
initialdir="data"
)
if not path:
return
self.account_manager.set_db_file(path)
self.integrity_tools.set_db_path(path)
self.audit_logger.log(f"Открыта база данных: {Path(path).name}")
self.refresh_users()
# --- Account ops ---
def refresh_users(self):
self.users_view.set_items(self.account_manager.list_users())
def add_user(self):
username = simpledialog.askstring(
"Добавить пользователя", "Введите имя пользователя:")
if not username:
return
password = simpledialog.askstring(
"Пароль", "Введите пароль:", show="*")
if not password:
return
salt = self.password_hasher.generate_salt()
password_hash = self.password_hasher.hash_password(password, salt)
self.account_manager.add_user(username, password_hash, salt)
self.audit_logger.log(f"Добавлен пользователь: {username}")
self.refresh_users()
def update_user(self):
username = self.users_view.get_selected()
if not username:
messagebox.showwarning("Изменить", "Выберите пользователя")
return
password = simpledialog.askstring(
"Новый пароль", f"Введите новый пароль для {username}:", show="*")
if not password:
return
salt = self.password_hasher.generate_salt()
password_hash = self.password_hasher.hash_password(password, salt)
if self.account_manager.update_user(username, password_hash, salt):
self.audit_logger.log(f"Обновлён пароль пользователя: {username}")
self.refresh_users()
def delete_user(self):
username = self.users_view.get_selected()
if not username:
messagebox.showwarning("Удалить", "Выберите пользователя")
return
if self.account_manager.delete_user(username):
self.audit_logger.log(f"Удалён пользователь: {username}")
self.refresh_users()
def run(self):
self.root.mainloop()
def close_db(self):
self.account_manager.set_db_file(None)
self.integrity_tools.set_db_path(None)
self.audit_logger.log("Закрыта база данных")
self.users_view.set_items([])
@@ -0,0 +1,21 @@
import tkinter as tk
class UserListView:
"""
Виджет списка пользователей.
"""
def __init__(self, master):
self.frame = tk.Frame(master)
self._listbox = tk.Listbox(self.frame, width=40)
self._listbox.pack(padx=10, pady=10)
def set_items(self, items: list[str]) -> None:
self._listbox.delete(0, tk.END)
for it in items:
self._listbox.insert(tk.END, it)
def get_selected(self) -> str | None:
sel = self._listbox.curselection()
return self._listbox.get(sel[0]) if sel else None
+17
View File
@@ -0,0 +1,17 @@
#!/bin/sh
# Создание виртуального окружения в .venv (если ещё нет)
if [ ! -d ".venv" ]; then
python3 -m venv .venv
fi
# Активация окружения
. .venv/bin/activate
# Обновление pip и установка зависимостей
pip install --upgrade pip
pip install -r requirements.txt
echo "Виртуальное окружение создано и зависимости установлены."
echo "Активируйте его командой: . .venv/bin/activate"
@@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['app.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='lab5-manager',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
@@ -0,0 +1,2 @@
cryptography>=42.0.0
pytest>=7.0
+14
View File
@@ -0,0 +1,14 @@
#!/bin/sh
# Проверка наличия окружения
if [ ! -d ".venv" ]; then
echo "Виртуальное окружение не найдено. Сначала выполните ./install.sh"
exit 1
fi
# Активация окружения
. .venv/bin/activate
# Запуск приложения
python app.py
+14
View File
@@ -0,0 +1,14 @@
#!/bin/sh
# Проверка наличия окружения
if [ ! -d ".venv" ]; then
echo "Виртуальное окружение не найдено. Сначала выполните ./install.sh"
exit 1
fi
# Активация окружения
. .venv/bin/activate
# Запуск тестов
pytest -v tests
@@ -0,0 +1,5 @@
import sys
import os
sys.path.insert(0, os.path.abspath(
os.path.join(os.path.dirname(__file__), "..")))
@@ -0,0 +1,23 @@
from core.account_manager import AccountManager
def test_crud_operations(tmp_path):
db_file = tmp_path / "accounts.json"
manager = AccountManager(db_file)
# Добавление
manager.add_user("alice", "hash1", "salt1")
assert "alice" in manager.list_users()
# Чтение
user = manager.get_user("alice")
assert user["hash"] == "hash1"
# Обновление
assert manager.update_user("alice", "hash2", "salt2")
user = manager.get_user("alice")
assert user["hash"] == "hash2"
# Удаление
assert manager.delete_user("alice")
assert "alice" not in manager.list_users()
@@ -0,0 +1,12 @@
from core.audit_logger import AuditLogger
def test_log_event(tmp_path):
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log("Test event")
content = log_file.read_text(encoding="utf-8")
assert "Test event" in content
assert "[" in content # должно быть время
@@ -0,0 +1,17 @@
from core.crypto_manager import CryptoManager
def test_encrypt_decrypt_file(tmp_path):
file = tmp_path / "plain.txt"
file.write_text("super secret", encoding="utf-8")
enc_file = tmp_path / "plain.enc"
dec_file = tmp_path / "plain.dec"
crypto = CryptoManager()
key = crypto.generate_key()
crypto.encrypt_file(file, key, enc_file)
crypto.decrypt_file(enc_file, key, dec_file)
assert dec_file.read_text(encoding="utf-8") == "super secret"
@@ -0,0 +1,17 @@
from pathlib import Path
from core.integrity_checker import IntegrityChecker
def test_calculate_and_verify_hash(tmp_path: Path):
file = tmp_path / "data.txt"
file.write_text("hello", encoding="utf-8")
hash_file = tmp_path / "data.hash"
checker = IntegrityChecker()
checker.save_hash(file, hash_file)
assert checker.verify_hash(file, hash_file)
# Нарушим целостность
file.write_text("changed", encoding="utf-8")
assert not checker.verify_hash(file, hash_file)
@@ -0,0 +1,12 @@
import pytest
from core.password_hasher import PasswordHasher
def test_hash_and_verify():
hasher = PasswordHasher()
salt = hasher.generate_salt()
pwd = "secret123"
hashed = hasher.hash_password(pwd, salt)
assert hasher.verify_password(pwd, salt, hashed)
assert not hasher.verify_password("wrong", salt, hashed)
+82
View File
@@ -0,0 +1,82 @@
import json
import sys
from cli.parse_args import parse_args
def _normalize_channel(ch: str) -> str:
"""Y/Cb/Cr в канонический вид."""
m = {"y": "Y", "cb": "Cb", "cr": "Cr"}
s = (ch or "").strip().lower()
return m.get(s, ch)
def _print_result(result):
if result is None:
return
if isinstance(result, (dict, list)):
print(json.dumps(result, ensure_ascii=False, indent=2))
else:
print(result)
def main() -> int:
args = parse_args()
try:
if args.cmd == "embed":
from modes.embed_mode import EmbedMode
runner = EmbedMode()
result = runner.run(
input_path=args.input,
label_path=args.label,
output_path=args.output,
channel=_normalize_channel(args.channel),
seq0=args.seq0,
)
elif args.cmd == "extract":
from modes.extract_mode import ExtractMode
runner = ExtractMode()
result = runner.run(
input_path=args.input,
output_path=args.output,
channel=_normalize_channel(args.channel),
seq0=args.seq0
)
elif args.cmd == "analyze":
from modes.analyze_quality import AnalyzeQuality
runner = AnalyzeQuality()
result = runner.run(
original_path=args.original,
stego_path=args.stego,
space=args.space,
metrics=args.metrics,
)
elif args.cmd == "generate":
from modes.generate_mode import GenerateMode
runner = GenerateMode()
# Ветка generate имеет вложенные подкоманды: args.gen in {"gradient","chess"}
result = runner.run(
kind=args.gen,
size=args.size,
channels=args.channels,
output_path=args.output,
tile=getattr(args, "tile", None),
)
else:
raise RuntimeError(f"Неизвестная команда: {args.cmd}")
_print_result(result)
return 0
except Exception as exc:
print(f"Ошибка: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
+3
View File
@@ -0,0 +1,3 @@
__all__ = ["parse_args"]
from .parse_args import parse_args
+147
View File
@@ -0,0 +1,147 @@
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
def parse_args(argv=None):
ap = ArgumentParser(
prog="app.py",
description="CLI для лабораторной №6: Стеганография в изображениях",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sub = ap.add_subparsers(dest="cmd", required=True)
# -------- Parent parsers --------
parent_channel_seq = ArgumentParser(add_help=False)
parent_channel_seq.add_argument(
"-c", "--channel",
choices=["Y", "Cb", "Cr", "y", "cb", "cr"],
default="Cb",
help="Канал для встраивания/извлечения: Y, Cb или Cr",
)
parent_channel_seq.add_argument(
"--seq0",
type=str,
default="01010101",
help="8-битная последовательность для бита 0 (seq1 считается инверсией seq0)",
)
parent_gen_common = ArgumentParser(add_help=False)
parent_gen_common.add_argument(
"--size",
type=int,
default=512,
help="Размер стороны квадратного изображения",
)
parent_gen_common.add_argument(
"--channels",
type=int,
choices=[1, 3],
default=3,
help="Количество каналов: 1 (grayscale) или 3 (RGB)",
)
parent_gen_common.add_argument(
"-o", "--output",
required=True,
help="Куда сохранить изображение",
)
# ---------------- EMBED ----------------
sp_embed = sub.add_parser(
"embed",
parents=[parent_channel_seq],
help="Встраивание метки в изображение-контейнер",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sp_embed.add_argument(
"-i", "--input",
required=True,
help="Путь к исходному изображению-контейнеру (512x512, RGB)",
)
sp_embed.add_argument(
"-l", "--label",
required=True,
help="Путь к бинарной метке (64x64)",
)
sp_embed.add_argument(
"-o", "--output",
required=True,
help="Куда сохранить стего-изображение",
)
# ---------------- EXTRACT ----------------
sp_extract = sub.add_parser(
"extract",
parents=[parent_channel_seq],
help="Извлечение метки из стего-изображения",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sp_extract.add_argument(
"-i", "--input",
required=True,
help="Путь к стего-изображению",
)
sp_extract.add_argument(
"-o", "--output",
required=True,
help="Куда сохранить восстановленную бинарную метку (64x64)",
)
# ---------------- ANALYZE ----------------
sp_analyze = sub.add_parser(
"analyze",
help="Сравнение изображений и метрик качества",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sp_analyze.add_argument(
"-a", "--original",
required=True,
help="Путь к исходному изображению-контейнеру",
)
sp_analyze.add_argument(
"-b", "--stego",
required=True,
help="Путь к стего-изображению",
)
sp_analyze.add_argument(
"--space",
choices=["rgb", "ycbcr"],
default="ycbcr",
help="Цветовое пространство для метрик",
)
sp_analyze.add_argument(
"--metrics",
nargs="+",
choices=["psnr"],
default=["psnr"],
help="Набор метрик",
)
# ---------------- GENERATE ----------------
sp_gen = sub.add_parser(
"generate",
help="Генерация изображений",
formatter_class=ArgumentDefaultsHelpFormatter,
)
gen_sub = sp_gen.add_subparsers(dest="gen", required=True)
gen_sub.add_parser(
"gradient",
parents=[parent_gen_common],
help="Сгенерировать градиент",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sp_chess = gen_sub.add_parser(
"chess",
parents=[parent_gen_common],
help="Сгенерировать шахматную доску",
formatter_class=ArgumentDefaultsHelpFormatter,
)
sp_chess.add_argument(
"--tile",
type=int,
default=32,
help="Размер квадрата (должен делить --size)",
)
return ap.parse_args(argv)
Binary file not shown.

After

Width:  |  Height:  |  Size: 2.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 118 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 444 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.8 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 114 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 133 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 959 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 667 B

+21
View File
@@ -0,0 +1,21 @@
__all__ = [
"Label64",
"Sequence8",
"ColorSpace",
"BlockGrid8",
"ChannelView",
"RgbImage512",
"PsnrMetric",
"YcbcrImage512",
"LsbRow8",
]
from .label64 import Label64
from .sequence8 import Sequence8
from .color_space import ColorSpace
from .block_grid8 import BlockGrid8
from .channel_view import ChannelView
from .rgb_image512 import RgbImage512
from .psnr_metric import PsnrMetric
from .ycbcr_image512 import YcbcrImage512
from .lsb_row8 import LsbRow8
@@ -0,0 +1,40 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class BlockGrid8:
"""Разбиение и сборка блоков 8×8."""
@staticmethod
def split(component: np.ndarray) -> np.ndarray:
"""Разбить 2D-компоненту на блоки (N,8,8)."""
if component.ndim != 2:
raise ValueError("Ожидается 2D-массив компоненты.")
h, w = component.shape
if h % 8 != 0 or w % 8 != 0:
raise ValueError("Размеры должны быть кратны 8.")
# Перестановка осей даёт (nH, nW, 8, 8) -> (N, 8, 8)
arr = np.asarray(component)
nH, nW = h // 8, w // 8
blocks = arr.reshape(nH, 8, nW, 8).transpose(
0, 2, 1, 3).reshape(nH * nW, 8, 8)
return blocks
@staticmethod
def merge(blocks: np.ndarray, height: int, width: int) -> np.ndarray:
"""Собрать 2D-компоненту из блоков (N,8,8)."""
if blocks.ndim != 3 or blocks.shape[1:] != (8, 8):
raise ValueError("Ожидается массив блоков формы (N,8,8).")
if height % 8 != 0 or width % 8 != 0:
raise ValueError("Размеры должны быть кратны 8.")
nH, nW = height // 8, width // 8
if blocks.shape[0] != nH * nW:
raise ValueError(
"Количество блоков не соответствует размеру выходной матрицы.")
# Обратная перестановка из (N,8,8) -> (nH, nW, 8, 8) -> (H, W)
arr = np.asarray(blocks)
comp = arr.reshape(nH, nW, 8, 8).transpose(
0, 2, 1, 3).reshape(height, width)
return comp
@@ -0,0 +1,39 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class ChannelView:
"""Доступ к компонентам Y, Cb, Cr в массиве YCbCr."""
_MAP = {"y": 0, "cb": 1, "cr": 2}
@staticmethod
def _idx(name: str) -> int:
"""Вернуть индекс канала."""
k = name.strip().lower()
if k not in ChannelView._MAP:
raise ValueError("Канал должен быть Y, Cb или Cr.")
return ChannelView._MAP[k]
@staticmethod
def get(ycbcr: np.ndarray, channel: str) -> np.ndarray:
"""Вернуть 2D-компоненту выбранного канала."""
if ycbcr.ndim != 3 or ycbcr.shape[-1] != 3:
raise ValueError("Ожидается массив HxWx3.")
i = ChannelView._idx(channel)
return ycbcr[..., i]
@staticmethod
def set(ycbcr: np.ndarray, channel: str, component: np.ndarray) -> np.ndarray:
"""Вернуть копию YCbCr с заменённым каналом."""
if ycbcr.ndim != 3 or ycbcr.shape[-1] != 3:
raise ValueError("Ожидается массив HxWx3.")
if component.shape != ycbcr.shape[:2]:
raise ValueError("Размер компоненты не совпадает с HxW.")
out = ycbcr.copy()
i = ChannelView._idx(channel)
# Короткая отсечка и приведение типа
out[..., i] = np.clip(component, 0, 255).astype(np.uint8)
return out
@@ -0,0 +1,44 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class ColorSpace:
"""Конвертация RGB↔YCbCr с отсечкой диапазона."""
# Матрица прямого преобразования (используются коэффициенты из задания)
_M = np.array([
[0.299, 0.587, 0.144],
[-0.168736, -0.331264, 0.5],
[0.5, -0.418688, -0.081312],
], dtype=np.float64)
_M_INV = np.linalg.inv(_M)
@staticmethod
def _clip_u8(arr: np.ndarray) -> np.ndarray:
"""Округлить и отсечь к uint8."""
return np.clip(np.rint(arr), 0, 255).astype(np.uint8)
@classmethod
def rgb_to_ycbcr(cls, rgb: np.ndarray) -> np.ndarray:
"""RGB uint8 -> YCbCr uint8."""
if rgb.ndim != 3 or rgb.shape[-1] != 3:
raise ValueError("Ожидается массив HxWx3.")
x = rgb.astype(np.float64).reshape(-1, 3)
ycbcr = x @ cls._M.T
ycbcr[:, 1:] += 128.0 # смещения Cb/Cr
ycbcr = ycbcr.reshape(rgb.shape)
return cls._clip_u8(ycbcr)
@classmethod
def ycbcr_to_rgb(cls, ycbcr: np.ndarray) -> np.ndarray:
"""YCbCr uint8 -> RGB uint8."""
if ycbcr.ndim != 3 or ycbcr.shape[-1] != 3:
raise ValueError("Ожидается массив HxWx3.")
x = ycbcr.astype(np.float64).reshape(-1, 3)
x[:, 1:] -= 128.0 # убрать смещения Cb/Cr
rgb = x @ cls._M_INV.T
rgb = rgb.reshape(ycbcr.shape)
return cls._clip_u8(rgb)
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable
import numpy as np
from PIL import Image
@dataclass(frozen=True)
class Label64:
"""Бинарная метка 64x64."""
data: np.ndarray # uint8, shape (64, 64), значения {0,1}
def __post_init__(self):
arr = np.asarray(self.data, dtype=np.uint8)
if arr.shape != (64, 64):
raise ValueError("Метка должна быть 64x64.")
# Нормализация к {0,1}
arr = (arr > 0).astype(np.uint8)
object.__setattr__(self, "data", arr)
@classmethod
def from_image(cls, path: str, threshold: int = 128) -> "Label64":
"""Создать из изображения 64x64."""
img = Image.open(path).convert("L")
if img.size != (64, 64):
raise ValueError("Изображение метки должно быть 64x64.")
arr = np.array(img, dtype=np.uint8)
arr = (arr >= threshold).astype(np.uint8)
return cls(arr)
@classmethod
def from_vector(cls, vec: Iterable[int]) -> "Label64":
"""Создать из вектора длиной 4096."""
v = np.fromiter(vec, dtype=np.uint8, count=4096)
if v.size != 4096:
raise ValueError("Длина вектора должна быть 4096.")
return cls(v.reshape(64, 64))
def to_vector(self) -> np.ndarray:
"""Вернуть вектор длиной 4096."""
return self.data.reshape(-1).astype(np.uint8)
def to_image(self, path: str) -> None:
"""Сохранить PNG 64x64."""
Image.fromarray(self.data * 255, mode="L").save(path)
+31
View File
@@ -0,0 +1,31 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class LsbRow8:
"""Встраивание/извлечение 8 бит в первый ряд блока 8×8 через LSB."""
@staticmethod
def embed(block: np.ndarray, bits: np.ndarray) -> np.ndarray:
"""Вернуть копию блока с вшитыми 8 битами."""
if block.shape != (8, 8):
raise ValueError("Нужен блок 8x8.")
b = np.asarray(block, dtype=np.uint8).copy()
v = np.asarray(bits, dtype=np.uint8).reshape(-1)
if v.size != 8 or not np.isin(v, [0, 1]).all():
raise ValueError("Нужно 8 бит 0/1.")
# Меняем только младшие биты первых 8 пикселей верхней строки
row = b[0, :8]
row = (row & 0xFE) | v
b[0, :8] = row
return b
@staticmethod
def extract(block: np.ndarray) -> np.ndarray:
"""Извлечь 8 бит из LSB первого ряда блока."""
if block.shape != (8, 8):
raise ValueError("Нужен блок 8x8.")
b = np.asarray(block, dtype=np.uint8)
return (b[0, :8] & 1).astype(np.uint8)
@@ -0,0 +1,22 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
@dataclass(frozen=True)
class PsnrMetric:
"""Расчёт PSNR для двух массивов одинаковой формы."""
max_value: float = 255.0
def psnr(self, a: np.ndarray, b: np.ndarray) -> float:
"""Вычислить PSNR."""
if a.shape != b.shape:
raise ValueError("Массивы должны иметь одинаковую форму.")
x = a.astype(np.float64)
y = b.astype(np.float64)
# MSE по всем элементам (каналы включены)
mse = np.mean((x - y) ** 2)
if mse == 0.0:
return float("inf")
return 20.0 * np.log10(self.max_value) - 10.0 * np.log10(mse)
@@ -0,0 +1,38 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
from PIL import Image
@dataclass(frozen=True)
class RgbImage512:
"""RGB-изображение 512x512."""
data: np.ndarray # uint8, shape (512, 512, 3)
def __post_init__(self):
arr = np.asarray(self.data, dtype=np.uint8)
if arr.shape != (512, 512, 3):
raise ValueError("Изображение должно быть 512x512x3 (RGB).")
object.__setattr__(self, "data", arr)
@classmethod
def from_file(cls, path: str) -> "RgbImage512":
"""Загрузить из файла."""
img = Image.open(path).convert("RGB")
if img.size != (512, 512):
raise ValueError("Размер изображения должен быть 512x512.")
return cls(np.array(img, dtype=np.uint8))
@classmethod
def from_array(cls, arr: np.ndarray) -> "RgbImage512":
"""Создать из массива 512x512x3."""
return cls(arr)
def to_array(self) -> np.ndarray:
"""Вернуть массив uint8 512x512x3."""
return self.data
def save(self, path: str) -> None:
"""Сохранить в файл."""
Image.fromarray(self.data, mode="RGB").save(path)
+55
View File
@@ -0,0 +1,55 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable
import numpy as np
@dataclass(frozen=True)
class Sequence8:
"""Пара 8-битных последовательностей: seq0 и инверсная seq1."""
_seq0: np.ndarray # uint8, shape (8,)
def __post_init__(self):
arr = np.asarray(self._seq0, dtype=np.uint8).reshape(-1)
if arr.size != 8:
raise ValueError("seq0 должна содержать 8 бит.")
if not np.isin(arr, [0, 1]).all():
raise ValueError("seq0 может состоять только из 0 и 1.")
object.__setattr__(self, "_seq0", arr)
@property
def seq0(self) -> np.ndarray:
"""Последовательность для бита 0."""
return self._seq0
@property
def seq1(self) -> np.ndarray:
"""Инверсия seq0 для бита 1."""
return 1 - self._seq0
def for_bit(self, bit: int) -> np.ndarray:
"""Вернуть последовательность по значению бита."""
if bit not in (0, 1):
raise ValueError("bit должен быть 0 или 1.")
return self._seq0 if bit == 0 else (1 - self._seq0)
@classmethod
def from_string(cls, bits: str) -> "Sequence8":
"""Создать из строки вида '01010101'."""
if len(bits) != 8 or any(ch not in "01" for ch in bits):
raise ValueError("Строка должна быть из 8 символов 0/1.")
arr = np.fromiter((int(ch) for ch in bits), dtype=np.uint8, count=8)
return cls(arr)
@classmethod
def from_iterable(cls, bits: Iterable[int]) -> "Sequence8":
"""Создать из итерируемого набора из 8 бит."""
arr = np.fromiter(bits, dtype=np.uint8, count=8)
if arr.size != 8:
raise ValueError("Нужно 8 бит.")
return cls(arr)
def __len__(self) -> int:
"""Длина последовательности."""
return 8
@@ -0,0 +1,50 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
# Локальные импорты внутри методов уменьшают риск циклов зависимостей
@dataclass(frozen=True)
class YcbcrImage512:
"""YCbCr-изображение 512x512."""
data: np.ndarray # uint8, shape (512, 512, 3)
def __post_init__(self):
arr = np.asarray(self.data, dtype=np.uint8)
if arr.shape != (512, 512, 3):
raise ValueError("Изображение должно быть 512x512x3 (YCbCr).")
object.__setattr__(self, "data", arr)
@classmethod
def from_rgb(cls, rgb_img) -> "YcbcrImage512":
"""Создать из RgbImage512."""
from .color_space import ColorSpace
from .rgb_image512 import RgbImage512
if not isinstance(rgb_img, RgbImage512):
raise TypeError("Ожидается RgbImage512.")
ycbcr = ColorSpace.rgb_to_ycbcr(rgb_img.to_array())
return cls(ycbcr)
def to_rgb(self):
"""Преобразовать в RgbImage512."""
from .color_space import ColorSpace
from .rgb_image512 import RgbImage512
rgb = ColorSpace.ycbcr_to_rgb(self.data)
return RgbImage512.from_array(rgb)
def get_channel(self, name: str) -> np.ndarray:
"""Вернуть 2D-компоненту канала."""
from .channel_view import ChannelView
return ChannelView.get(self.data, name)
def with_channel(self, name: str, component: np.ndarray) -> "YcbcrImage512":
"""Вернуть копию с заменённым каналом."""
from .channel_view import ChannelView
new_ycbcr = ChannelView.set(self.data, name, component)
return YcbcrImage512(new_ycbcr)
def to_array(self) -> np.ndarray:
"""Вернуть массив uint8 512x512x3."""
return self.data
@@ -0,0 +1,5 @@
__all__ = ["TestImageGenerator"]
from .chessboard_generator import ChessboardGenerator
from .gradient_generator import GradientGenerator
@@ -0,0 +1,33 @@
from __future__ import annotations
import numpy as np
class ChessboardGenerator:
"""Шахматная доска произвольного размера и каналов."""
@staticmethod
def make(size: int = 512, channels: int = 3, tile: int = 32) -> np.ndarray:
"""Создать шахматку size×size; для RGB каждый квадрат случайного цвета."""
if size <= 0:
raise ValueError("size должен быть > 0.")
if channels not in (1, 3):
raise ValueError("channels должен быть 1 или 3.")
if tile <= 0 or size % tile != 0:
raise ValueError("tile должен быть > 0 и делить size без остатка.")
n = size // tile # число клеток по стороне
if channels == 1:
base = (np.add.outer(np.arange(n), np.arange(n)) % 2).astype(np.uint8) * 255
return np.kron(base, np.ones((tile, tile), dtype=np.uint8))
# RGB: случайный цвет на каждую клетку (uint8)
rng = np.random.default_rng()
colors = rng.integers(0, 256, size=(n, n, 3), dtype=np.uint8)
# Апсемплинг каждой компоненты по плитке
r = np.kron(colors[:, :, 0], np.ones((tile, tile), dtype=np.uint8))
g = np.kron(colors[:, :, 1], np.ones((tile, tile), dtype=np.uint8))
b = np.kron(colors[:, :, 2], np.ones((tile, tile), dtype=np.uint8))
return np.stack([r, g, b], axis=-1)
@@ -0,0 +1,26 @@
from __future__ import annotations
import numpy as np
class GradientGenerator:
"""Градиентное изображение произвольного размера и каналов."""
@staticmethod
def make(size: int = 512, channels: int = 3) -> np.ndarray:
"""Создать квадратный градиент size×size для 1 или 3 каналов."""
if not isinstance(size, int) or size <= 0:
raise ValueError("size должен быть положительным целым.")
if channels not in (1, 3):
raise ValueError("channels должен быть 1 или 3.")
x = np.linspace(0.0, 255.0, num=size, dtype=np.float64)
h = np.tile(x, (size, 1)) # горизонтальный градиент
v = np.tile(x[:, None], (1, size)) # вертикальный градиент
d = (0.5 * (h + v)) # диагональный градиент
if channels == 1:
return d.astype(np.uint8)
rgb = np.stack([h, v, d], axis=2).astype(np.uint8) # R=горизонталь, G=вертикаль, B=диагональ
return rgb
+6
View File
@@ -0,0 +1,6 @@
__all__ = ["EmbedMode", "ExtractMode", "AnalyzeQuality", "GenerateMode"]
from .embed_mode import EmbedMode
from .extract_mode import ExtractMode
from .analyze_quality import AnalyzeQuality
from .generate_mode import GenerateMode
@@ -0,0 +1,41 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List
import numpy as np
from entities.rgb_image512 import RgbImage512
from entities.color_space import ColorSpace
from entities.psnr_metric import PsnrMetric
@dataclass
class AnalyzeQuality:
"""CLI-режим расчёта метрик качества."""
def run(
self,
original_path: str,
stego_path: str,
space: str = "ycbcr",
metrics: List[str] = None,
) -> Dict[str, float]:
"""Посчитать метрики между изображениями."""
if metrics is None:
metrics = ["psnr"]
metrics = [m.lower() for m in metrics]
if any(m != "psnr" for m in metrics):
raise ValueError("Поддерживается только метрика PSNR.")
orig = RgbImage512.from_file(original_path).to_array()
steg = RgbImage512.from_file(stego_path).to_array()
if space.lower() == "ycbcr":
a = ColorSpace.rgb_to_ycbcr(orig)
b = ColorSpace.rgb_to_ycbcr(steg)
elif space.lower() == "rgb":
a, b = orig, steg
else:
raise ValueError("space должен быть 'rgb' или 'ycbcr'.")
psnr = PsnrMetric().psnr(a, b)
return {"psnr": float(psnr)}

Some files were not shown because too many files have changed in this diff Show More