tempus-ddb 0.5.4

Fail-closed policy gate for AI agent actions: signed policies, single-use permits, receipts
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

import json
import sqlite3

import pytest
from tempus_ddb import TempusDDB, gen_keys


def test_tamper_delete_middle_record(tmp_path):
    """Test that deleting a record from the middle of the chain breaks validation."""
    keyfile = tmp_path / "keys.json"
    db_path = tmp_path / "test.db"
    gen_keys(str(keyfile))
    db = TempusDDB(str(db_path), str(keyfile))

    db.record(json.dumps({"step": 1}), "{}", genesis=True)
    db.record(json.dumps({"step": 2}), "{}", genesis=False)
    db.record(json.dumps({"step": 3}), "{}", genesis=False)

    conn = sqlite3.connect(str(db_path))
    try:
        # Get the second record
        cur = conn.cursor()
        cur.execute("SELECT id FROM decisions ORDER BY timestamp ASC LIMIT 1 OFFSET 1")
        mid_hash = cur.fetchone()[0]
        # Delete it
        conn.execute("DELETE FROM decisions WHERE id = ?", (mid_hash,))
        conn.commit()
    finally:
        conn.close()

    with pytest.raises(Exception) as exc:
        db.validate()
    assert (
        "invalid" in str(exc.value).lower()
        or "error" in str(exc.value).lower()
        or "mismatch" in str(exc.value).lower()
        or "parent" in str(exc.value).lower()
    )


def test_tamper_modify_payload(tmp_path):
    """Test that modifying a payload after recording is detected."""
    keyfile = tmp_path / "keys.json"
    db_path = tmp_path / "test.db"
    gen_keys(str(keyfile))
    db = TempusDDB(str(db_path), str(keyfile))

    db.record(json.dumps({"amount": 100}), "{}", genesis=True)

    conn = sqlite3.connect(str(db_path))
    try:
        conn.execute(
            "UPDATE decisions SET payload = ?", (json.dumps({"amount": 1000}),)
        )
        conn.commit()
    finally:
        conn.close()

    with pytest.raises(Exception) as exc:
        db.validate()
    assert (
        "invalid" in str(exc.value).lower()
        or "error" in str(exc.value).lower()
        or "mismatch" in str(exc.value).lower()
    )


def test_json_canonicalization_whitespace(tmp_path):
    """Test that whitespace in JSON doesn't change the underlying canonical payload."""
    keyfile = tmp_path / "keys.json"
    db_path = tmp_path / "test.db"
    gen_keys(str(keyfile))
    db = TempusDDB(str(db_path), str(keyfile))

    # Insert with lots of whitespace
    payload = '{   "a"  :   1 , \n "b" : 2  }'
    db.record(payload, "{}", genesis=True)

    # Should be valid
    db.validate()

    conn = sqlite3.connect(str(db_path))
    try:
        cur = conn.cursor()
        cur.execute("SELECT payload FROM decisions LIMIT 1")
        stored_payload = cur.fetchone()[0]
        # It should be canonicalized by the Rust core (or at least strictly parsed)
        assert json.loads(stored_payload) == {"a": 1, "b": 2}
    finally:
        conn.close()


def test_replay_attack_prevention(tmp_path):
    """Test that inserting the exact same payload in quick succession works as distinct events."""
    keyfile = tmp_path / "keys.json"
    db_path = tmp_path / "test.db"
    gen_keys(str(keyfile))
    db = TempusDDB(str(db_path), str(keyfile))

    db.record('{"event":"login"}', "{}", genesis=True)
    db.record('{"event":"login"}', "{}", genesis=False)

    # The chain should still be valid, they should have different hashes because the parent hash differs
    val = db.validate()
    assert val is not None


def test_fork_behavior_second_genesis(tmp_path):
    """Test that attempting a second genesis fails."""
    keyfile = tmp_path / "keys.json"
    db_path = tmp_path / "test.db"
    gen_keys(str(keyfile))
    db = TempusDDB(str(db_path), str(keyfile))

    db.record('{"step":1}', "{}", genesis=True)

    with pytest.raises(Exception):
        db.record('{"step":2}', "{}", genesis=True)