tempus-ddb 0.5.3

Local-first B2A security gate with tamper-evident execution 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

"""Tests for Monotonic Checkpoints and Append-Only EventStream verification."""

import json
import os
import tempfile
import time

from tempus_ddb import TempusDDB, gen_keys


def _setup_gate(tmpdir, tenant_id="tenant-checkpoint-test"):
    gate_db = os.path.join(tmpdir, "gate.db")
    gate_key = os.path.join(tmpdir, "gate.keys.json")
    agent_key = os.path.join(tmpdir, "agent.keys.json")
    exec_key = os.path.join(tmpdir, "exec.keys.json")

    gen_keys(gate_key)
    gen_keys(agent_key)
    gen_keys(exec_key)

    gate = TempusDDB(gate_db, gate_key)

    with open(gate_key, "r", encoding="utf-8") as f:
        gate_id = json.load(f)["public_key"]
    with open(agent_key, "r", encoding="utf-8") as f:
        agent_id = json.load(f)["public_key"]
    with open(exec_key, "r", encoding="utf-8") as f:
        exec_id = json.load(f)["public_key"]

    gate.register_agent(
        gate_id, "gate-root", json.dumps({"can_delegate": True, "tenant_id": tenant_id})
    )
    gate.register_agent(agent_id, "agent-coder", json.dumps({"tenant_id": tenant_id}))
    gate.register_agent(exec_id, "exec-runner", json.dumps({"tenant_id": tenant_id}))

    return gate, gate_id, agent_id, exec_id, gate_key, agent_key, exec_key, tenant_id


def test_event_stream_hash_chain():
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        gate, gate_id, agent_id, exec_id, gate_key, agent_key, exec_key, tenant_id = (
            _setup_gate(tmpdir)
        )

        # Export event stream
        stream_str = gate.export_event_stream(tenant_id, 1, 100)
        events = json.loads(stream_str)

        assert (
            len(events) >= 3
        )  # gate root, agent-coder, exec-runner registrations + policy
        assert events[0]["sequence_number"] == 1
        assert (
            events[0]["prev_event_hash"]
            == "0000000000000000000000000000000000000000000000000000000000000000"
        )

        # Check contiguous hash links
        for i in range(1, len(events)):
            assert events[i]["sequence_number"] == events[i - 1]["sequence_number"] + 1
            assert events[i]["prev_event_hash"] == events[i - 1]["event_digest"]


def test_create_and_verify_checkpoint():
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        gate, gate_id, agent_id, exec_id, gate_key, agent_key, exec_key, tenant_id = (
            _setup_gate(tmpdir)
        )

        # Submit an authorized action
        intent = {
            "schema_version": "tempus.action-intent.v1",
            "tenant_id": tenant_id,
            "agent_id": agent_id,
            "idempotency_key": "chk-action-01",
            "action_type": "github.create_issue",
            "resource": "elbuilder77/tempus-ddb",
            "requested_at": int(time.time() * 1_000_000),
            "input": {"title": "Checkpoints in 0.5"},
        }
        auth_resp_str = gate.request_action(json.dumps(intent), agent_key, 60)
        auth_resp = json.loads(auth_resp_str)
        auth_id = auth_resp["authorization"]["authorization_id"]

        # Commit outcome
        outcome = {
            "schema_version": "tempus.action-outcome.v1",
            "authorization_id": auth_id,
            "action_id": auth_resp["authorization"]["action_id"],
            "executor_id": exec_id,
            "status": "SUCCEEDED",
            "output": {
                "html_url": "https://github.com/elbuilder77/tempus-ddb/issues/101"
            },
            "completed_at": int(time.time() * 1_000_000),
        }
        gate.commit_outcome(auth_id, json.dumps(outcome), exec_key)

        # Create monotonic checkpoint
        chk_str = gate.create_checkpoint(tenant_id)
        chk = json.loads(chk_str)

        assert chk["schema_version"] == "tempus.checkpoint.v1"
        assert chk["checkpoint_sequence"] == 1
        assert chk["total_events"] >= 5
        assert chk["signature"]

        # Export stream and verify offline
        stream_str = gate.export_event_stream(tenant_id, 1, 1000)
        verify_result_str = gate.verify_checkpoint_stream(chk_str, stream_str)
        verify_result = json.loads(verify_result_str)

        assert verify_result["status"] == "VERIFIED"
        assert verify_result["reason_code"] == "CHECKPOINT_VALID"
        assert verify_result["events_verified"] == chk["total_events"]


def test_adversarial_tamper_event_detection():
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        gate, gate_id, agent_id, exec_id, gate_key, agent_key, exec_key, tenant_id = (
            _setup_gate(tmpdir)
        )

        chk_str = gate.create_checkpoint(tenant_id)
        stream_str = gate.export_event_stream(tenant_id, 1, 100)
        events = json.loads(stream_str)

        # Tamper payload_hash of an event
        events[1]["payload_hash"] = (
            "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
        )

        verify_result_str = gate.verify_checkpoint_stream(chk_str, json.dumps(events))
        verify_result = json.loads(verify_result_str)

        assert verify_result["status"] == "INVALID"
        assert verify_result["reason_code"] in {
            "ERR_EVENT_TAMPERED",
            "ERR_STREAM_HASH_MISMATCH",
        }


def test_adversarial_signature_tamper_detection():
    with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
        gate, gate_id, agent_id, exec_id, gate_key, agent_key, exec_key, tenant_id = (
            _setup_gate(tmpdir)
        )

        chk_str = gate.create_checkpoint(tenant_id)
        chk = json.loads(chk_str)
        # Invalidate signature
        chk["signature"] = "00" * 64
        stream_str = gate.export_event_stream(tenant_id, 1, 100)

        verify_result_str = gate.verify_checkpoint_stream(json.dumps(chk), stream_str)
        verify_result = json.loads(verify_result_str)

        assert verify_result["status"] == "INVALID"
        assert verify_result["reason_code"] == "ERR_SIGNATURE_INVALID"