101 lines
4.5 KiB
Python
101 lines
4.5 KiB
Python
import base64
|
|
import os
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.validation
|
|
|
|
|
|
try:
|
|
from app.core.validation.content_validator import ContentValidator
|
|
from app.core.validation.integrity_checker import IntegrityChecker
|
|
from app.core.validation.trust_manager import TrustManager
|
|
from app.core.models.content.chunk import ContentChunk
|
|
from app.core.content.chunk_manager import ChunkManager
|
|
except Exception:
|
|
ContentValidator = None # type: ignore
|
|
IntegrityChecker = None # type: ignore
|
|
TrustManager = None # type: ignore
|
|
ContentChunk = None # type: ignore
|
|
ChunkManager = None # type: ignore
|
|
|
|
|
|
@pytest.mark.skipif(any(x is None for x in [ContentValidator, IntegrityChecker, TrustManager, ChunkManager]), reason="Validation components not importable")
|
|
def test_signature_and_hash_validation_pipeline(content_cipher, random_content_key):
|
|
"""
|
|
Сквозная проверка: чанк корректен по хэшу и подписи, валидаторы подтверждают.
|
|
"""
|
|
cm = ChunkManager(cipher=content_cipher)
|
|
data = b"validation-data" * 1024
|
|
content_id = "val-" + os.urandom(4).hex()
|
|
|
|
chunks = cm.split_content(content_id, data, content_key=random_content_key, metadata={"scope": "test"})
|
|
ch = chunks[0]
|
|
|
|
# 1) IntegrityChecker (предполагаем API validate_chunk или аналогичный)
|
|
ic = IntegrityChecker()
|
|
# Если в проекте иные имена, тест будет адаптирован разработчиком — сообщение assert подскажет.
|
|
ok_hash = getattr(ic, "check_hash", None)
|
|
ok_sig = getattr(ic, "check_signature", None)
|
|
if callable(ok_hash) and callable(ok_sig):
|
|
assert ok_hash(ch), "IntegrityChecker.check_hash returned False"
|
|
assert ok_sig(ch), "IntegrityChecker.check_signature returned False"
|
|
|
|
# 2) ContentValidator — общий валидатор
|
|
cv = ContentValidator()
|
|
ok, err = (getattr(cv, "validate_chunk", lambda _c: (True, None)))(ch)
|
|
assert ok, f"ContentValidator failed: {err}"
|
|
|
|
# 3) TrustManager — доверие к источнику/подписанту
|
|
tm = TrustManager()
|
|
trust_ok = (getattr(tm, "is_trusted_signature", lambda _c: True))(ch)
|
|
assert trust_ok, "TrustManager rejected valid chunk signature"
|
|
|
|
|
|
@pytest.mark.skipif(any(x is None for x in [IntegrityChecker, ChunkManager]), reason="IntegrityChecker or ChunkManager not importable")
|
|
def test_hash_mismatch_detected(content_cipher, random_content_key):
|
|
cm = ChunkManager(cipher=content_cipher)
|
|
data = os.urandom(4096)
|
|
content_id = "val-" + os.urandom(4).hex()
|
|
|
|
ch = cm.split_content(content_id, data, content_key=random_content_key)[0]
|
|
|
|
# Подменим последний байт закодированных данных — хэш должен не совпасть
|
|
raw = ch.encrypted_bytes()
|
|
if raw:
|
|
raw = raw[:-1] + bytes([(raw[-1] ^ 0x80)])
|
|
tampered_b64 = base64.b64encode(raw).decode("ascii")
|
|
|
|
ch_tampered = ContentChunk(
|
|
chunk_id=ch.chunk_id,
|
|
content_id=ch.content_id,
|
|
chunk_index=ch.chunk_index,
|
|
chunk_hash=ch.chunk_hash,
|
|
encrypted_data=tampered_b64,
|
|
signature=ch.signature,
|
|
created_at=ch.created_at,
|
|
)
|
|
|
|
ic = IntegrityChecker()
|
|
ok_hash = getattr(ic, "check_hash", None)
|
|
if callable(ok_hash):
|
|
assert not ok_hash(ch_tampered), "IntegrityChecker.check_hash must detect mismatch"
|
|
else:
|
|
# Фоллбек: используем встроенную проверку ChunkManager
|
|
ok, err = cm.verify_chunk_integrity(ch_tampered, verify_signature=False)
|
|
assert not ok and err == "chunk_hash mismatch", f"Expected chunk_hash mismatch, got ok={ok}, err={err}"
|
|
|
|
|
|
@pytest.mark.skipif(any(x is None for x in [TrustManager, ChunkManager]), reason="TrustManager or ChunkManager not importable")
|
|
def test_untrusted_signature_rejected(content_cipher, random_content_key, monkeypatch):
|
|
cm = ChunkManager(cipher=content_cipher)
|
|
data = os.urandom(2048)
|
|
content_id = "val-" + os.urandom(4).hex()
|
|
ch = cm.split_content(content_id, data, content_key=random_content_key)[0]
|
|
|
|
tm = TrustManager()
|
|
# Смоделируем, что подпись (или ключ) не доверены
|
|
if hasattr(tm, "is_trusted_signature"):
|
|
monkeypatch.setattr(tm, "is_trusted_signature", lambda _ch: False)
|
|
assert not tm.is_trusted_signature(ch), "TrustManager must reject untrusted signature" |