HDK-audit / tests /test_integration.py

Initial release: code, docs, tests (CC BY-NC 4.0)

977ccde 24 days ago

7.66 kB

"""Integration tests — end-to-end workflows, tampering detection, performance."""
import asyncio
import json
import threading
import time
from datetime import datetime, timezone
from unittest.mock import patch

from hdk import HDK, HDKConfig
from hdk.certificate import CertificateGenerator
from hdk.core.canary import CanaryLayer
from hdk.core.dag import DAGStorage
from hdk.core.hash_engine import HashEngine
from hdk.hedera.anchor import OfflineAnchorService
from hdk.models import EventType, HDKNode, NodeLevel

from tests.conftest import TEST_KEY


def test_offline_to_online():
    """100 LLM calls offline, then flush pending anchors."""
    dag = DAGStorage(":memory:")
    engine = HashEngine()
    canary = CanaryLayer(dag, key=TEST_KEY)
    canary.initialize()

    # Add 100 nodes
    for i in range(100):
        content_hash = engine.hash_content(f"content_{i}")
        self_hash = engine.hash_with_parents(content_hash, [])
        node = HDKNode(
            id=f"node-{i}", level=NodeLevel.QUERY,
            content_hash=content_hash, parent_chain=[], self_hash=self_hash,
            event_type=EventType.LLM_QUERY, model="gpt-4",
        )
        canary.weave_into_node(node)
        node.node_hash = engine.create_node_hash(node)
        dag.add_node(node)

    service = OfflineAnchorService(dag, canary)
    assert service.get_pending_count() == 0

    # Anchor offline
    anchor = asyncio.run(service.submit_anchor(force=True))
    assert anchor is not None
    assert anchor.node_count == 100
    assert service.get_pending_count() == 1  # pending (no Hedera)

    dag.close()


def test_tampering_detection_counter():
    """Attacker modifies read counter in SQLite → canary detects."""
    dag = DAGStorage(":memory:")
    canary = CanaryLayer(dag, key=TEST_KEY)
    canary.initialize()

    # Normal reads
    canary.authorized_read()
    canary.authorized_read()
    checkpoint = canary.create_checkpoint()

    # Tamper with counter
    dag._conn.execute("UPDATE canary_state SET read_counter = 999 WHERE id = 1")
    dag._conn.commit()

    # Reload and verify
    canary2 = CanaryLayer(dag, key=TEST_KEY)
    assert canary2.verify(checkpoint["commitment"], checkpoint["counter"]) is False
    dag.close()


def test_tampering_detection_node_hash():
    """Attacker modifies content_hash in SQLite → verify_integrity catches."""
    dag = DAGStorage(":memory:")
    engine = HashEngine()

    content_hash = engine.hash_content("original content")
    self_hash = engine.hash_with_parents(content_hash, [])
    node = HDKNode(
        id="node-1", level=NodeLevel.EVENT,
        content_hash=content_hash, parent_chain=[], self_hash=self_hash,
        event_type=EventType.CUSTOM,
    )
    node.node_hash = engine.create_node_hash(node)
    dag.add_node(node)

    # Tamper
    dag._conn.execute(
        "UPDATE nodes SET content_hash = 'TAMPERED' WHERE id = 'node-1'"
    )
    dag._conn.commit()

    valid, errors = dag.verify_integrity()
    assert valid is False
    assert len(errors) > 0
    dag.close()


def test_tampering_detection_merkle_root():
    """Attacker modifies merkle_root in anchor → verification fails."""
    dag = DAGStorage(":memory:")
    engine = HashEngine()
    canary = CanaryLayer(dag, key=TEST_KEY)
    canary.initialize()

    # Add nodes
    for i in range(5):
        ch = engine.hash_content(f"c_{i}")
        sh = engine.hash_with_parents(ch, [])
        n = HDKNode(
            id=f"n-{i}", level=NodeLevel.EVENT,
            content_hash=ch, parent_chain=[], self_hash=sh,
            event_type=EventType.CUSTOM,
        )
        canary.weave_into_node(n)
        n.node_hash = engine.create_node_hash(n)
        dag.add_node(n)

    service = OfflineAnchorService(dag, canary)
    anchor = asyncio.run(service.submit_anchor(force=True))
    original_root = anchor.merkle_root

    # Tamper merkle root
    dag._conn.execute(
        "UPDATE anchors SET merkle_root = 'FAKE_ROOT' WHERE anchor_sequence = 0"
    )
    dag._conn.commit()

    # Read tampered anchor
    tampered = dag.get_last_anchor()
    assert tampered.merkle_root != original_root
    assert tampered.merkle_root == "FAKE_ROOT"
    valid, errors = dag.verify_integrity()
    assert valid is False
    assert any("merkle_root mismatch" in e for e in errors)
    dag.close()


def test_certificate_roundtrip():
    """Generate → JSON → parse → verify."""
    hdk = HDK(config=HDKConfig(db_path=":memory:"))

    hdk.record_event(EventType.LLM_QUERY, "prompt")
    hdk.record_event(EventType.LLM_RESPONSE, "response")

    gen = CertificateGenerator(hdk._dag, hdk._engine)
    doc = "My final document"
    cert = gen.generate(doc, "author_123")

    # Serialize
    json_str = gen.to_json(cert)
    data = json.loads(json_str)

    # Verify fields survive roundtrip
    assert data["document_hash"] == cert.document_hash
    assert data["author_hash"] == cert.author_hash
    assert data["merkle_root"] == cert.merkle_root

    # Verify against document
    result = gen.verify_certificate(cert, doc)
    assert result["document_hash_valid"] is True
    assert result["overall"] is True

    hdk.close()


def test_large_dag_performance():
    """Add 10,000 nodes. merkle_root < 1s, verify < 5s, export < 2s."""
    dag = DAGStorage(":memory:")
    engine = HashEngine()

    # Add 10k nodes
    for i in range(10_000):
        ch = engine.hash_content(f"content_{i}")
        sh = engine.hash_with_parents(ch, [])
        n = HDKNode(
            id=f"n-{i}", level=NodeLevel.EVENT,
            content_hash=ch, parent_chain=[], self_hash=sh,
            event_type=EventType.LLM_QUERY,
        )
        n.node_hash = engine.create_node_hash(n)
        dag.add_node(n)

    assert dag.get_node_count() == 10_000

    # Merkle root < 1s
    t0 = time.time()
    root = dag.compute_current_merkle_root()
    t_merkle = time.time() - t0
    assert len(root) == 64
    assert t_merkle < 1.0, f"merkle_root took {t_merkle:.2f}s"

    # Verify < 5s
    t0 = time.time()
    valid, errors = dag.verify_integrity()
    t_verify = time.time() - t0
    assert valid is True
    assert t_verify < 5.0, f"verify took {t_verify:.2f}s"

    # Export < 2s
    t0 = time.time()
    exported = dag.export_dag()
    t_export = time.time() - t0
    assert t_export < 2.0, f"export took {t_export:.2f}s"

    data = json.loads(exported)
    assert len(data["nodes"]) == 10_000

    dag.close()


def test_concurrent_writes():
    """Multiple threads writing to DAG simultaneously — no corruption."""
    dag = DAGStorage(":memory:")
    engine = HashEngine()
    errors = []

    def writer(thread_id: int, count: int):
        try:
            for i in range(count):
                ch = engine.hash_content(f"t{thread_id}_c{i}")
                sh = engine.hash_with_parents(ch, [])
                n = HDKNode(
                    id=f"t{thread_id}-{i}", level=NodeLevel.EVENT,
                    content_hash=ch, parent_chain=[], self_hash=sh,
                    event_type=EventType.CUSTOM,
                )
                n.node_hash = engine.create_node_hash(n)
                dag.add_node(n)
        except Exception as e:
            errors.append(str(e))

    threads = [threading.Thread(target=writer, args=(t, 50)) for t in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # All nodes written (SQLite serializes writes, so this should work)
    total = dag.get_node_count()
    assert total == 200, f"Expected 200, got {total}. Errors: {errors}"

    # Integrity still valid
    valid, errs = dag.verify_integrity()
    assert valid is True

    dag.close()

	"""Integration tests — end-to-end workflows, tampering detection, performance."""
	import asyncio
	import json
	import threading
	import time
	from datetime import datetime, timezone
	from unittest.mock import patch

	from hdk import HDK, HDKConfig
	from hdk.certificate import CertificateGenerator
	from hdk.core.canary import CanaryLayer
	from hdk.core.dag import DAGStorage
	from hdk.core.hash_engine import HashEngine
	from hdk.hedera.anchor import OfflineAnchorService
	from hdk.models import EventType, HDKNode, NodeLevel

	from tests.conftest import TEST_KEY


	def test_offline_to_online():
	"""100 LLM calls offline, then flush pending anchors."""
	dag = DAGStorage(":memory:")
	engine = HashEngine()
	canary = CanaryLayer(dag, key=TEST_KEY)
	canary.initialize()

	# Add 100 nodes
	for i in range(100):
	content_hash = engine.hash_content(f"content_{i}")
	self_hash = engine.hash_with_parents(content_hash, [])
	node = HDKNode(
	id=f"node-{i}", level=NodeLevel.QUERY,
	content_hash=content_hash, parent_chain=[], self_hash=self_hash,
	event_type=EventType.LLM_QUERY, model="gpt-4",
	)
	canary.weave_into_node(node)
	node.node_hash = engine.create_node_hash(node)
	dag.add_node(node)

	service = OfflineAnchorService(dag, canary)
	assert service.get_pending_count() == 0

	# Anchor offline
	anchor = asyncio.run(service.submit_anchor(force=True))
	assert anchor is not None
	assert anchor.node_count == 100
	assert service.get_pending_count() == 1 # pending (no Hedera)

	dag.close()


	def test_tampering_detection_counter():
	"""Attacker modifies read counter in SQLite → canary detects."""
	dag = DAGStorage(":memory:")
	canary = CanaryLayer(dag, key=TEST_KEY)
	canary.initialize()

	# Normal reads
	canary.authorized_read()
	canary.authorized_read()
	checkpoint = canary.create_checkpoint()

	# Tamper with counter
	dag._conn.execute("UPDATE canary_state SET read_counter = 999 WHERE id = 1")
	dag._conn.commit()

	# Reload and verify
	canary2 = CanaryLayer(dag, key=TEST_KEY)
	assert canary2.verify(checkpoint["commitment"], checkpoint["counter"]) is False
	dag.close()


	def test_tampering_detection_node_hash():
	"""Attacker modifies content_hash in SQLite → verify_integrity catches."""
	dag = DAGStorage(":memory:")
	engine = HashEngine()

	content_hash = engine.hash_content("original content")
	self_hash = engine.hash_with_parents(content_hash, [])
	node = HDKNode(
	id="node-1", level=NodeLevel.EVENT,
	content_hash=content_hash, parent_chain=[], self_hash=self_hash,
	event_type=EventType.CUSTOM,
	)
	node.node_hash = engine.create_node_hash(node)
	dag.add_node(node)

	# Tamper
	dag._conn.execute(
	"UPDATE nodes SET content_hash = 'TAMPERED' WHERE id = 'node-1'"
	)
	dag._conn.commit()

	valid, errors = dag.verify_integrity()
	assert valid is False
	assert len(errors) > 0
	dag.close()


	def test_tampering_detection_merkle_root():
	"""Attacker modifies merkle_root in anchor → verification fails."""
	dag = DAGStorage(":memory:")
	engine = HashEngine()
	canary = CanaryLayer(dag, key=TEST_KEY)
	canary.initialize()

	# Add nodes
	for i in range(5):
	ch = engine.hash_content(f"c_{i}")
	sh = engine.hash_with_parents(ch, [])
	n = HDKNode(
	id=f"n-{i}", level=NodeLevel.EVENT,
	content_hash=ch, parent_chain=[], self_hash=sh,
	event_type=EventType.CUSTOM,
	)
	canary.weave_into_node(n)
	n.node_hash = engine.create_node_hash(n)
	dag.add_node(n)

	service = OfflineAnchorService(dag, canary)
	anchor = asyncio.run(service.submit_anchor(force=True))
	original_root = anchor.merkle_root

	# Tamper merkle root
	dag._conn.execute(
	"UPDATE anchors SET merkle_root = 'FAKE_ROOT' WHERE anchor_sequence = 0"
	)
	dag._conn.commit()

	# Read tampered anchor
	tampered = dag.get_last_anchor()
	assert tampered.merkle_root != original_root
	assert tampered.merkle_root == "FAKE_ROOT"
	valid, errors = dag.verify_integrity()
	assert valid is False
	assert any("merkle_root mismatch" in e for e in errors)
	dag.close()


	def test_certificate_roundtrip():
	"""Generate → JSON → parse → verify."""
	hdk = HDK(config=HDKConfig(db_path=":memory:"))

	hdk.record_event(EventType.LLM_QUERY, "prompt")
	hdk.record_event(EventType.LLM_RESPONSE, "response")

	gen = CertificateGenerator(hdk._dag, hdk._engine)
	doc = "My final document"
	cert = gen.generate(doc, "author_123")

	# Serialize
	json_str = gen.to_json(cert)
	data = json.loads(json_str)

	# Verify fields survive roundtrip
	assert data["document_hash"] == cert.document_hash
	assert data["author_hash"] == cert.author_hash
	assert data["merkle_root"] == cert.merkle_root

	# Verify against document
	result = gen.verify_certificate(cert, doc)
	assert result["document_hash_valid"] is True
	assert result["overall"] is True

	hdk.close()


	def test_large_dag_performance():
	"""Add 10,000 nodes. merkle_root < 1s, verify < 5s, export < 2s."""
	dag = DAGStorage(":memory:")
	engine = HashEngine()

	# Add 10k nodes
	for i in range(10_000):
	ch = engine.hash_content(f"content_{i}")
	sh = engine.hash_with_parents(ch, [])
	n = HDKNode(
	id=f"n-{i}", level=NodeLevel.EVENT,
	content_hash=ch, parent_chain=[], self_hash=sh,
	event_type=EventType.LLM_QUERY,
	)
	n.node_hash = engine.create_node_hash(n)
	dag.add_node(n)

	assert dag.get_node_count() == 10_000

	# Merkle root < 1s
	t0 = time.time()
	root = dag.compute_current_merkle_root()
	t_merkle = time.time() - t0
	assert len(root) == 64
	assert t_merkle < 1.0, f"merkle_root took {t_merkle:.2f}s"

	# Verify < 5s
	t0 = time.time()
	valid, errors = dag.verify_integrity()
	t_verify = time.time() - t0
	assert valid is True
	assert t_verify < 5.0, f"verify took {t_verify:.2f}s"

	# Export < 2s
	t0 = time.time()
	exported = dag.export_dag()
	t_export = time.time() - t0
	assert t_export < 2.0, f"export took {t_export:.2f}s"

	data = json.loads(exported)
	assert len(data["nodes"]) == 10_000

	dag.close()


	def test_concurrent_writes():
	"""Multiple threads writing to DAG simultaneously — no corruption."""
	dag = DAGStorage(":memory:")
	engine = HashEngine()
	errors = []

	def writer(thread_id: int, count: int):
	try:
	for i in range(count):
	ch = engine.hash_content(f"t{thread_id}_c{i}")
	sh = engine.hash_with_parents(ch, [])
	n = HDKNode(
	id=f"t{thread_id}-{i}", level=NodeLevel.EVENT,
	content_hash=ch, parent_chain=[], self_hash=sh,
	event_type=EventType.CUSTOM,
	)
	n.node_hash = engine.create_node_hash(n)
	dag.add_node(n)
	except Exception as e:
	errors.append(str(e))

	threads = [threading.Thread(target=writer, args=(t, 50)) for t in range(4)]
	for t in threads:
	t.start()
	for t in threads:
	t.join()

	# All nodes written (SQLite serializes writes, so this should work)
	total = dag.get_node_count()
	assert total == 200, f"Expected 200, got {total}. Errors: {errors}"

	# Integrity still valid
	valid, errs = dag.verify_integrity()
	assert valid is True

	dag.close()