1#!/usr/bin/env python3
2import argparse
3import hashlib
4import html
5import http.client
6import http.server
7import json
8import os
9from pathlib import Path
10import socket
11import sqlite3
12import ssl
13import struct
14import subprocess
15import tempfile
16import threading
17import time
18import urllib.parse
19
20
21def main():
22 parser = argparse.ArgumentParser()
23 parser.add_argument('--binary', type=Path, required=True)
24 args = parser.parse_args()
25 with tempfile.TemporaryDirectory(prefix='shale-attribution-') as temporary:
26 root = Path(temporary).resolve()
27 (root / 'data').mkdir()
28 cert, key = root / 'cert.pem', root / 'key.pem'
29 subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(key),
30 '-out', str(cert), '-days', '1', '-subj', '/CN=localhost',
31 '-addext', 'subjectAltName=DNS:localhost,IP:127.0.0.1', '-addext', 'basicConstraints=critical,CA:FALSE'],
32 check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
33 state = {'caller_write': True, 'status': 'todo', 'posts': [], 'comments': []}
34
35 def page(body, kind='page-issue'):
36 return "<meta name='astheno.shale.repo.name' content='chat'><body id='" + kind + "'>" + body + '</body>'
37
38 def issue():
39 comments = ''.join("<li class=comment id=c%d><div class=n-card__header><a href=/~%s>~%s</a></div><div class=markdown>%s</div></li>" %
40 (i, author, author, html.escape(comment)) for i, (author, comment) in enumerate(state['comments'], 1))
41 return '<h1><span>#1</span><span>Fixture</span></h1><dl class=sidebar><dd><span class=issuestatus-' + state['status'] + '>' + state['status'] + '</span></dd></dl><ul class=timeline>' + comments + '</ul>'
42
43 class Backend(http.server.BaseHTTPRequestHandler):
44 def log_message(self, *args):
45 pass
46
47 def reply(self, code, body='', **headers):
48 encoded = body.encode()
49 self.send_response(code)
50 for name, value in headers.items():
51 self.send_header(name, value)
52 self.send_header('Content-Length', str(len(encoded)))
53 self.end_headers()
54 self.wfile.write(encoded)
55
56 def do_GET(self):
57 actor = False
58 if self.path == '/-/settings':
59 self.reply(200, '<body id=page-user-settings><kbd>' + 'human' + '</kbd></body>')
60 return
61 if self.path == '/':
62 self.reply(200, '<body id=page-index><div class=grid-container><table><tbody><tr><td><a href=/chat/>chat</a></td><td>Fixture</td><td></td></tr><tr><td><a href=/other/>other</a></td><td>Other</td><td></td></tr></tbody></table></div></body>')
63 return
64 permitted = state['caller_write']
65 csrf = 'actor-token' if actor else 'caller-token'
66 if self.path == '/chat/issues/labels':
67 self.reply(200, page("<div class=grid-container><table><tbody><tr><td><a class=issuelabel data-label-id=75 data-color='#123456'>Bug</a></td><td>Fixture</td><td>2</td></tr></tbody></table></div>", 'page-labels'))
68 return
69 if self.path == '/chat/issues/new':
70 form = '<form method=post><input name=title><input type=hidden name=csrf_token value=' + csrf + '></form>' if permitted else ''
71 self.reply(200, page(form, 'page-new-issue'))
72 return
73 if self.path == '/chat/issues/1':
74 form = '<form method=post><input type=hidden name=t value=comment><input type=hidden name=csrf_token value=' + csrf + '></form>' if permitted else ''
75 self.reply(200, page(issue() + form))
76 return
77 self.reply(404)
78
79 def do_POST(self):
80 state['posts'].append(self.path)
81 self.reply(500, 'Native HTTP writes must not be used')
82
83 backend = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Backend)
84 context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
85 context.load_cert_chain(cert, key)
86 backend.socket = context.wrap_socket(backend.socket, server_side=True)
87 thread = threading.Thread(target=backend.serve_forever, daemon=True)
88 thread.start()
89 with socket.socket() as available:
90 available.bind(('127.0.0.1', 0))
91 port = available.getsockname()[1]
92 host_calls = []
93 host_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
94 host_socket.bind(str(root / 'host.sock'))
95 host_socket.listen()
96 host_socket.settimeout(.1)
97 host_done = threading.Event()
98
99 def host_requests():
100 while not host_done.is_set():
101 try:
102 connection, _ = host_socket.accept()
103 except socket.timeout:
104 continue
105 with connection:
106 request = json.loads(connection.makefile('rb').readline())
107 if request['operation'] != 'shale.write':
108 result = json.dumps({'value': {}}).encode()
109 try:
110 connection.sendall(struct.pack('>I', len(result)) + result)
111 except BrokenPipeError:
112 pass
113 continue
114 host_calls.append(request)
115 fields = request['arguments']
116 if request['action'] == 'set_issue_status':
117 state['status'] = fields['status']
118 if request['action'] == 'create_issue':
119 state['status'] = fields['status']
120 if request['action'] in ('create_issue', 'comment_issue'):
121 state['comments'].append(('ai-human-' + fields['model'], fields.get('description', fields.get('comment', ''))))
122 result = json.dumps({'value': {'actor': '~ai-human-' + fields['model'], 'id': 1}}).encode()
123 try:
124 connection.sendall(struct.pack('>I', len(result)) + result)
125 except BrokenPipeError:
126 pass
127
128 host_thread = threading.Thread(target=host_requests, daemon=True)
129 host_thread.start()
130 origin = 'https://globe.studio.test'
131 environment = {**os.environ, 'PORT': str(port), 'STUDIO_LISTEN_ADDRESS': '127.0.0.1',
132 'STUDIO_DATA_DIR': str(root / 'data'), 'STUDIO_PUBLIC_ORIGIN': origin,
133 'STUDIO_AUTH_RP_ID': 'studio.test', 'STUDIO_FILE_ORIGIN': 'https://file.studio.test',
134 'STUDIO_SHALE_URL': 'https://localhost:' + str(backend.server_port),
135 'STUDIO_CA_BUNDLE': str(cert), 'STUDIO_REPO': str(Path(__file__).resolve().parent.parent),
136 'STUDIO_AUTH_REQUIRED': '0', 'STUDIO_HOST_SOCKET': str(root / 'host.sock')}
137 for name in ('STUDIO_PROXY_TOKEN_FILE', 'STUDIO_INTERNAL_URL', 'STUDIO_YT_STATE', 'STUDIO_INDEX_POOL'):
138 environment.pop(name, None)
139 with (root / 'server.log').open('wb') as log:
140 server = subprocess.Popen([str(args.binary.resolve())], env=environment, stdout=log, stderr=log)
141 try:
142 deadline = time.monotonic() + 20
143 while True:
144 try:
145 with socket.create_connection(('127.0.0.1', port), timeout=.1):
146 break
147 except OSError:
148 assert server.poll() is None, (root / 'server.log').read_text()[-2000:]
149 assert time.monotonic() < deadline
150 time.sleep(.05)
151 with sqlite3.connect(root / 'data/accounts.sqlite') as db:
152 for name in ('human', 'ai'):
153 db.execute('INSERT INTO users(id,profile) VALUES (?,?)', (name, json.dumps({'username': name, 'enabled': True, 'requiredActions': []})))
154 resource = origin + '/mcp/shale'
155 grant = {'id': 'fixture', 'user': 'human', 'resource': resource, 'scopes': ['shale:read', 'shale:write'], 'resources': ['chat']}
156 with sqlite3.connect(root / 'data/connections.sqlite') as db:
157 for name, value in [('grant:fixture', grant), ('access:' + hashlib.sha256(b'fixture').hexdigest(), {'grant': 'fixture', 'resource': resource}),
158 ('shale-session:human', {'origin': 'https://localhost:' + str(backend.server_port) + '/', 'session': 'human'}),
159 ('shale-session:ai', {'origin': 'https://localhost:' + str(backend.server_port) + '/', 'session': 'actor'})]:
160 db.execute('INSERT INTO records VALUES (?,?,0)', (name, json.dumps(value)))
161
162 def call(name, fields, error=False):
163 connection = http.client.HTTPConnection('127.0.0.1', port, timeout=15)
164 connection.request('POST', '/mcp/shale', json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call', 'params': {'name': name, 'arguments': fields}}),
165 {'Host': 'globe.studio.test', 'Authorization': 'Bearer fixture', 'Accept': 'application/json, text/event-stream', 'Content-Type': 'application/json'})
166 response = connection.getresponse()
167 body = response.read()
168 connection.close()
169 assert response.status == 200, (response.status, body)
170 result = json.loads(body)['result']
171 assert bool(result.get('isError', False)) == error, result
172 return result
173
174 created = call('create_issue', {'repository': 'chat', 'title': 'Fixture', 'description': 'Description', 'model': 'claude-opus-5.5', 'status': 'todo'})
175 assert created['structuredContent']['issue']['comments'][0]['author'] == '~ai-human-claude-opus-5.5'
176 assert created['structuredContent']['issue']['status'] == 'todo'
177 commented = call('comment_issue', {'repository': 'chat', 'id': 1, 'comment': 'Comment', 'model': 'gpt-6.1-sol'})
178 assert commented['structuredContent']['issue']['comments'][-1]['author'] == '~ai-human-gpt-6.1-sol'
179 assert host_calls[0]['arguments']['description'] == 'Description\n\nAssisted-by: claude-opus-5.5'
180 assert host_calls[1]['arguments']['comment'] == 'Comment\n\nAssisted-by: gpt-6.1-sol'
181 for body in ('x' * 200000, '\x01' * 240000, '😀' * 60000):
182 call('comment_issue', {'repository': 'chat', 'id': 1, 'comment': body, 'model': 'gpt-6.1-sol'})
183 assert host_calls[-1]['arguments']['comment'].startswith(body)
184 call('comment_issue', {'repository': 'chat', 'id': 1, 'comment': 'x' * 262144, 'model': 'gpt-6.1-sol'}, error=True)
185 call('create_issue', {'repository': 'chat', 'title': 'Question', 'model': 'gpt-6.1-sol'})
186 assert state['status'] == 'needs_triage'
187 for body, model in [('Body\n\nAssisted-by: claude-opus-5.5', 'gpt-6.1-sol'), ('```\nBody', 'gpt-6.1-sol'), ('Body', '<model-id>')]:
188 call('comment_issue', {'repository': 'chat', 'id': 1, 'comment': body, 'model': model}, error=True)
189 for fields in ({'title': 'Missing model'}, {'title': 'Invalid state', 'model': 'gpt-6.1-sol', 'status': 'done'}):
190 call('create_issue', {'repository': 'chat', **fields}, error=True)
191 state['caller_write'] = False
192 changed = call('set_issue_status', {'repository': 'chat', 'id': 1, 'status': 'done', 'model': 'gpt-6.1-sol'})
193 assert changed['structuredContent']['issue']['status'] == 'done'
194 assert host_calls[-1] == {'operation': 'shale.write', 'grant': 'fixture', 'repository': 'chat', 'action': 'set_issue_status', 'arguments': {'id': 1, 'status': 'done', 'model': 'gpt-6.1-sol'}}
195 catalog = call('list_labels', {'repository': 'chat'})['structuredContent']
196 assert catalog['labels'][0]['id'] == 75 and catalog['labels'][0]['openIssues'] == 2
197 for name, fields in [('create_label', {'name': 'Bug', 'description': '', 'color': '#123456'}),
198 ('update_label', {'label_id': 75, 'name': 'Defect', 'description': '', 'color': '#abcdef'}),
199 ('set_issue_labels', {'id': 1, 'labels': [75]}), ('delete_label', {'label_id': 75})]:
200 call(name, {'repository': 'chat', 'model': 'gpt-6.1-sol', **fields})
201 assert host_calls[-1] == {'operation': 'shale.write', 'grant': 'fixture', 'repository': 'chat', 'action': name, 'arguments': {'model': 'gpt-6.1-sol', **fields}}
202 call(name, {'repository': 'chat', **fields}, error=True)
203 count = len(host_calls)
204 call('create_label', {'repository': 'other', 'model': 'gpt-6.1-sol', 'name': 'REFUSED', 'description': '', 'color': '#123456'}, error=True)
205 with sqlite3.connect(root / 'data/connections.sqlite') as db:
206 db.execute('UPDATE records SET value=? WHERE key=?', (json.dumps({**grant, 'scopes': ['shale:read']}), 'grant:fixture'))
207 call('set_issue_labels', {'repository': 'chat', 'id': 1, 'labels': [], 'model': 'gpt-6.1-sol'}, error=True)
208 call('set_issue_status', {'repository': 'chat', 'id': 1, 'status': 'todo', 'model': 'gpt-6.1-sol'}, error=True)
209 assert len(host_calls) == count
210 assert state['posts'] == [], 'A bridge write used a native user session'
211 print(json.dumps({'durable_model_arguments': True, 'triage_default_and_todo': True, 'trailers_match_model': True, 'missing_models_refused': True, 'scope_and_readonly_enforced': True, 'no_native_session_write': True}))
212
213 finally:
214 server.terminate()
215 server.wait(timeout=10)
216 backend.shutdown()
217 backend.server_close()
218 host_done.set()
219 host_thread.join(timeout=1)
220 host_socket.close()
221
222
223if __name__ == '__main__':
224 main()