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
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

#!/usr/bin/env python3
"""Build a deterministic, credential-free runtime from reviewed local archives.

No network, native execution, root build or runtime installation is performed.
Pinned archive links are materialized as regular files; output has no symlinks.
"""
import argparse
import gzip
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import struct
import tarfile

ROOT = Path(__file__).resolve().parents[1]
SOURCES = ROOT / 'docs/operator-packets/e579/runtime/sources.json'
MAX_FILE = 256 * 1024 * 1024
MAX_TOTAL = 1024 * 1024 * 1024


def digest(path):
    result = hashlib.sha256()
    with path.open('rb') as stream:
        for chunk in iter(lambda: stream.read(1048576), b''):
            result.update(chunk)
    return result.hexdigest()


def safe(name):
    return bool(name) and all(part not in ('', '.', '..') for part in name.split('/')) and bool(re.fullmatch('[A-Za-z0-9/._-]+', name))


def archive(path, expected):
    if path.is_symlink() or not path.is_file() or digest(path) != expected:
        raise ValueError('reviewed archive digest mismatch')
    value = tarfile.open(path, 'r:gz')
    members = value.getmembers()
    if len(members) > 16384 or len({m.name for m in members}) != len(members):
        value.close()
        raise ValueError('archive duplicate or unbounded inventory')
    for member in members:
        if (not safe(member.name.rstrip('/')) or member.size > MAX_FILE
                or not (member.isfile() or member.isdir() or member.issym() or member.islnk())):
            value.close()
            raise ValueError('unsafe archive member')
    return value


def member_bytes(source, name):
    member = source.getmember(name)
    seen = set()
    while member.issym() or member.islnk():
        if member.name in seen or len(seen) >= 16:
            raise ValueError('archive link cycle')
        seen.add(member.name)
        target = (PurePosixPath(member.name).parent / member.linkname if member.issym()
                  else PurePosixPath(member.linkname))
        # Resolve only inside the pinned archive, never on the host filesystem.
        parts = []
        if target.is_absolute():
            raise ValueError('absolute archive link')
        for part in target.parts:
            if part == '..':
                if not parts:
                    raise ValueError('escaping archive link')
                parts.pop()
            elif part != '.':
                parts.append(part)
        member = source.getmember('/'.join(parts))
    if not member.isfile() or member.size > MAX_FILE:
        raise ValueError('bounded regular archive member required')
    with source.extractfile(member) as stream:
        raw = stream.read(MAX_FILE + 1)
    if len(raw) != member.size or len(raw) > MAX_FILE:
        raise ValueError('archive member changed or exceeded bound')
    return raw


def arm64(raw):
    # Native little-endian Mach-O64 with CPU_TYPE_ARM64, not a venv/link/script.
    if len(raw) < 8 or struct.unpack('<II', raw[:8]) != (0xfeedfacf, 0x0100000c):
        raise ValueError('native arm64 Mach-O interpreter/client required')


def build(python_archive, app_archive, seeds_archive, output, sources=SOURCES):
    if os.geteuid() == 0:
        raise ValueError('build as unprivileged user; never build or download as root')
    pins = json.loads(sources.read_text())
    inputs = [(python_archive, pins['python']['sha256']), (app_archive, pins['shantytown']['sha256']),
              (seeds_archive, pins['seeds']['sha256'])]
    # Verify all inputs before creating an output. Never extractall foreign paths.
    archives = []
    try:
        for path, pin in inputs:
            archives.append(archive(path, pin))
        py, app, sd = archives
        files = {}
        files['runtime/bin/python3'] = member_bytes(py, 'python/bin/python3')
        arm64(files['runtime/bin/python3'])
        for member in py.getmembers():
            name = member.name
            if not name.startswith('python/lib/') or member.isdir():
                continue
            relative = name.removeprefix('python/')
            # No pip/site hooks, bytecode, tests or developer config in a root runtime.
            if any(part in {'site-packages', '__pycache__', 'test', 'tests'} for part in PurePosixPath(relative).parts):
                continue
            files['runtime/' + relative] = member_bytes(py, name)
        prefix = 'shantytown-' + pins['shantytown']['commit'] + '/'
        for member in app.getmembers():
            if member.isfile() and member.name.startswith(prefix + 'shantytown/'):
                relative = member.name.removeprefix(prefix)
                if any(part in {'__pycache__', 'tests', '.git'} for part in PurePosixPath(relative).parts):
                    continue
                files['runtime/app/' + relative] = member_bytes(app, member.name)
        files['launcher'] = member_bytes(app, prefix + 'scripts/st-job-launcher')
        sd_name = 'sd-seeds-ai-v' + pins['seeds']['version'] + '-aarch64-apple-darwin/sd'
        files['runtime/bin/sd'] = member_bytes(sd, sd_name)
        arm64(files['runtime/bin/sd'])
        # Keep shipped licenses as data in an allowed library subtree.
        files['runtime/lib/licenses/shantytown-LICENSE'] = member_bytes(app, prefix + 'LICENSE')
        files['runtime/lib/licenses/seeds-LICENSE'] = member_bytes(sd, str(PurePosixPath(sd_name).parent / 'LICENSE'))
        if (not files['launcher'].startswith(b'#!/opt/shantytown-job-launcher/bin/python3 -BIS\n')
                or len(files) > 4096 or sum(map(len, files.values())) > MAX_TOTAL
                or any(not safe(name) for name in files)):
            raise ValueError('runtime inventory or launcher outside delivery contract')
        for required in ['runtime/app/shantytown/job_launch.py', 'runtime/app/shantytown/job_metadata.py',
                         'runtime/lib/python3.12/os.py', 'runtime/lib/python3.12/encodings/__init__.py']:
            if required not in files:
                raise ValueError('incomplete runtime')
        output.mkdir(mode=0o755)  # Exclusive; partial output is retained on later failure.
        manifest = {'version': 1, 'platform': 'macos', 'files': {name: hashlib.sha256(raw).hexdigest()
                                                               for name, raw in sorted(files.items())}}
        for name, raw in sorted(files.items()):
            path = output / name
            path.parent.mkdir(parents=True, exist_ok=True)
            with path.open('xb') as stream:
                stream.write(raw)
            path.chmod(0o555 if name == 'launcher' or name.startswith('runtime/bin/') else 0o444)
        manifest_path = output / 'manifest.json'
        manifest_path.write_text(json.dumps(manifest, sort_keys=True, indent=2) + '\n')
        manifest_path.chmod(0o444)
        pin = digest(manifest_path)
        for path in sorted(output.rglob('*'), reverse=True):
            if path.is_dir():
                path.chmod(0o555)
        output.chmod(0o555)
        return pin
    finally:
        for source in archives:
            source.close()


def pack(directory, target):
    # Stable mtimes/owners/modes/order and gzip header; no local path in the tar.
    with target.open('xb') as stream, gzip.GzipFile(fileobj=stream, mode='wb', filename='', mtime=0) as zipped:
        with tarfile.open(fileobj=zipped, mode='w', format=tarfile.PAX_FORMAT) as output:
            for path in [directory, *sorted(directory.rglob('*'))]:
                name = directory.name + ('/' + str(path.relative_to(directory)) if path != directory else '')
                info = output.gettarinfo(str(path), arcname=name)
                info.uid = info.gid = info.mtime = 0
                info.uname = info.gname = ''
                if path.is_file():
                    with path.open('rb') as source:
                        output.addfile(info, source)
                else:
                    output.addfile(info)
    target.with_name(target.name + '.sha256').write_text(digest(target) + '  ' + target.name + '\n')


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--python-archive', type=Path, required=True)
    parser.add_argument('--shantytown-archive', type=Path, required=True)
    parser.add_argument('--seeds-archive', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    pin = build(args.python_archive, args.shantytown_archive, args.seeds_archive, args.output)
    target = args.output.with_name(args.output.name + '.tar.gz')
    pack(args.output, target)
    print(json.dumps({'manifest_sha256': pin, 'archive': target.name, 'archive_sha256': digest(target)}))


if __name__ == '__main__':
    main()