tambo 0.14.0

Install and prove the whole AI knowledge-tool kit
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

#!/usr/bin/env python3
"""Bounded hosted readback; build policy and credentials stay on trusted host."""

import hashlib
import importlib.util
import json
from pathlib import Path
import re
import subprocess
import sys
import tarfile
import tempfile
import time

# Release source must stay clean when modules load.
sys.dont_write_bytecode = True

spec = importlib.util.spec_from_file_location("publisher", Path(__file__).with_name("publish_repository_share.py"))
publisher = importlib.util.module_from_spec(spec)
spec.loader.exec_module(publisher)


def verify(folder, tag, revision):
    if not re.fullmatch(r"v\d+\.\d+\.\d+(?:[.-][A-Za-z0-9.-]+)?", tag) or not re.fullmatch("[0-9a-f]{40}", revision):
        raise ValueError("invalid release identity")
    archive = folder / ("tambo-" + tag + "-repository.pendant.tar.gz")
    sha = hashlib.sha256(archive.read_bytes()).hexdigest()
    if archive.with_name(archive.name + ".sha256").read_text() != sha + "  " + archive.name + "\n":
        raise ValueError("archive checksum mismatch")
    proof = json.loads(archive.with_name(archive.name + ".provenance.json").read_text())
    expected = {
        "schema": "quipu.repository-share-producer/v1",
        "source_repository": publisher.REPOSITORY,
        "source_revision": revision,
        "release_tag": tag,
        "archive_sha256": sha,
        "destination": "outward",
    }
    if any(proof.get(k) != v for k, v in expected.items()):
        raise ValueError("repository/source/tag/provenance mismatch")
    for key in ("quipu_sha256", "quipu_server_sha256", "bobbin_sha256"):
        if not re.fullmatch("[0-9a-f]{64}", proof.get(key, "")):
            raise ValueError("missing engine/indexer digest")
    if not re.fullmatch("[0-9a-f]{40}", proof.get("engine_source_revision", "")):
        raise ValueError("missing engine source pin")
    with tempfile.TemporaryDirectory(prefix="verify-pendant-") as temp:
        stage = Path(temp)
        with tarfile.open(archive) as tar:
            members = tar.getmembers()
            if len(members) != len({m.name for m in members}) or any(
                not m.isfile() or m.name not in publisher.MEMBERS for m in members
            ):
                raise ValueError("unexpected, duplicate, or unsafe archive member")
            if any(m.name == "alignments.ttl" and not 0 < m.size <= publisher.MAX_ALIGNMENT_BYTES for m in members):
                raise ValueError("alignment archive member exceeds ancillary bound")
            # Every member is a flat allowlisted regular file; no path, link or device extraction.
            for member in members:
                (stage / member.name).write_bytes(tar.extractfile(member).read())
        publisher.check_share(stage)
        if proof.get("alignments_sha256") != publisher.alignment_digest(stage):
            raise ValueError("alignment provenance differs from verified carrier")
    return expected


def wait(tag, revision, attempts=80, pause=time.sleep, run=subprocess.run):
    if not re.fullmatch(r"v\d+\.\d+\.\d+(?:[.-][A-Za-z0-9.-]+)?", tag) or not re.fullmatch("[0-9a-f]{40}", revision):
        raise ValueError("invalid release identity before lookup")
    stem = "tambo-" + tag + "-repository.pendant.tar.gz"
    last_read = "not attempted"
    for number in range(attempts):
        with tempfile.TemporaryDirectory(prefix="repository-pendant-readback-") as temp:
            folder = Path(temp)
            try:
                result = run(
                    [
                        "gh",
                        "release",
                        "download",
                        tag,
                        "--repo",
                        publisher.REPOSITORY,
                        "--dir",
                        temp,
                        "--pattern",
                        stem,
                        "--pattern",
                        stem + ".sha256",
                        "--pattern",
                        stem + ".provenance.json",
                    ],
                    capture_output=True,
                    text=True,
                    timeout=45,
                )
                last_read = "exit=" + str(result.returncode)
                if result.returncode == 0:
                    return verify(folder, tag, revision)
            except subprocess.TimeoutExpired:
                last_read = "read timeout"
        if number + 1 < attempts:
            pause(30)
    raise ValueError("repository pendant unverified after bounded trusted-producer wait; " + last_read)


if __name__ == "__main__":
    if len(sys.argv) != 3:
        raise SystemExit("usage: verify-repository-share.py <release-tag> <exact-release-sha>")
    print(json.dumps(wait(sys.argv[1], sys.argv[2])))