1#!/usr/bin/env python3
2import argparse
3import base64
4from contextlib import closing, contextmanager
5import copy
6from html.parser import HTMLParser
7import http.client
8import http.cookiejar
9import http.server
10import importlib
11import json
12import os
13from pathlib import Path
14import re
15import sqlite3
16import ssl
17import socket
18import subprocess
19import sys
20import tempfile
21import threading
22import time
23import urllib.error
24import urllib.parse
25import urllib.request
26import uuid
27
28
29class Page(HTMLParser):
30 def __init__(self, text):
31 super().__init__()
32 self.forms = []
33 self.links = []
34 self.codes = []
35 self.form = None
36 self.code = None
37 self.select = None
38 self.feed(text)
39
40 def handle_starttag(self, tag, attrs):
41 attrs = dict(attrs)
42 if tag == "form":
43 self.form = {**attrs, "fields": []}
44 self.forms.append(self.form)
45 if tag in {"input", "select", "textarea", "button"} and self.form is not None:
46 self.form["fields"].append(attrs)
47 if tag == "select":
48 self.select = attrs
49 attrs["options"] = []
50 if tag == "option" and self.select is not None:
51 self.select["options"].append(attrs)
52 if tag == "a":
53 self.links.append(attrs.get("href", ""))
54 if tag in {"code", "pre", "textarea"}:
55 self.code = ""
56
57 def handle_data(self, data):
58 if self.code is not None:
59 self.code += data
60
61 def handle_endtag(self, tag):
62 if tag == "select":
63 self.select = None
64 if tag == "form":
65 self.form = None
66 if tag in {"code", "pre", "textarea"} and self.code is not None:
67 self.codes.append(self.code)
68 self.code = None
69
70
71@contextmanager
72def instance(keycloak, marker, image, traces_endpoint, traces_env, *, host="localhost", dashboard_port=None):
73 container = "studio-shale-backend-test-" + marker
74 client_name = "shale-fixture-" + marker
75 client_secret = uuid.uuid4().hex + uuid.uuid4().hex
76 client_id = None
77 caddy = None
78 proxy = None
79
80 def run(*argv):
81 try:
82 return subprocess.check_output(argv, text=True, stderr=subprocess.PIPE).strip()
83 except subprocess.CalledProcessError as error:
84 raise AssertionError(f"{argv[0]} failed ({error.returncode})") from None
85
86 with tempfile.TemporaryDirectory(prefix="studio-shale-backend-", dir="/run") as temporary:
87 root = Path(temporary)
88 with socket.socket() as reservation:
89 reservation.bind(("127.0.0.1", 0))
90 backend_port = reservation.getsockname()[1]
91 with socket.socket() as reservation:
92 for port in range(20000, 32001):
93 try:
94 reservation.bind(("127.0.0.1", port))
95 break
96 except OSError:
97 continue
98 else:
99 raise AssertionError("no fixture HTTPS port available")
100 origin = "https://" + host + ":" + str(port)
101 try:
102 keycloak.request("/admin/realms/master/clients", "POST", {
103 "clientId": client_name, "protocol": "openid-connect", "publicClient": False,
104 "standardFlowEnabled": True, "secret": client_secret,
105 "redirectUris": [origin + "/-/callback"], "webOrigins": [origin],
106 })
107 clients = keycloak.request("/admin/realms/master/clients?clientId=" + client_name)
108 assert len(clients) == 1
109 client_id = clients[0]["id"]
110 authority = Path("/var/lib/caddy/.local/share/caddy/pki/authorities/local")
111 run("openssl", "req", "-new", "-newkey", "rsa:2048", "-nodes", "-keyout", str(root / "key.pem"),
112 "-out", str(root / "request.pem"), "-subj", "/CN=" + host,
113 "-addext", "subjectAltName=DNS:" + host + ",IP:127.0.0.1")
114 run("openssl", "x509", "-req", "-in", str(root / "request.pem"), "-CA", str(authority / "root.crt"),
115 "-CAkey", str(authority / "root.key"), "-set_serial", "0x" + marker,
116 "-out", str(root / "cert.pem"), "-days", "1", "-copy_extensions", "copyall")
117 config = root / "Caddyfile"
118 bridge = importlib.import_module("router").shale_mcp_routes(dashboard_port) if dashboard_port else []
119 upstream_port = backend_port
120 if dashboard_port:
121 drop = root / "drop-next-write"
122
123 class Proxy(http.server.BaseHTTPRequestHandler):
124 def log_message(self, *args):
125 pass
126
127 def do_GET(self):
128 with closing(http.client.HTTPConnection("127.0.0.1", backend_port, timeout=20)) as upstream:
129 upstream.request(self.command, self.path,
130 self.rfile.read(int(self.headers.get("Content-Length", 0))), dict(self.headers))
131 response = upstream.getresponse()
132 body = response.read()
133 if self.command == "POST" and drop.exists() and drop.read_text() == self.path:
134 drop.unlink()
135 self.close_connection = True
136 self.connection.shutdown(socket.SHUT_RDWR)
137 return
138 self.send_response(response.status)
139 for name, value in response.getheaders():
140 if name.lower() not in {"connection", "transfer-encoding", "content-length"}:
141 self.send_header(name, value)
142 self.send_header("Content-Length", str(len(body)))
143 self.end_headers()
144 self.wfile.write(body)
145
146 do_POST = do_GET
147
148 proxy = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Proxy)
149 proxy_thread = threading.Thread(target=proxy.serve_forever, daemon=True)
150 proxy_thread.start()
151 upstream_port = proxy.server_port
152 config.write_text("{\n admin off\n auto_https disable_redirects\n skip_install_trust\n}\n" + origin +
153 " {\n tls " + str(root / "cert.pem") + " " + str(root / "key.pem") +
154 "\n" + "\n".join(bridge) + "\n handle {\n reverse_proxy 127.0.0.1:" + str(upstream_port) + "\n }\n}\n")
155 binary = (Path("/proc") / run("systemctl", "show", "-P", "MainPID", "caddy") / "exe").resolve(strict=True)
156 with (root / "caddy.log").open("w") as output:
157 caddy = subprocess.Popen([str(binary), "run", "--config", str(config), "--adapter", "caddyfile"],
158 stdout=output, stderr=output,
159 env={**os.environ, "XDG_DATA_HOME": temporary, "XDG_CONFIG_HOME": temporary})
160 volumes = []
161 for directory, target in [("data", "/data"), ("owned", "/repositories_owned"), ("mirrors", "/repositories_mirrors"), ("cron", "/etc/crontabs")]:
162 path = root / directory
163 path.mkdir()
164 os.chown(path, 65534, 65534)
165 volumes.append("--volume=" + str(path) + ":" + target + ":rw")
166 run("podman", "run", "-d", "--name=" + container, "--pull=never", "--user=65534:65534",
167 "--read-only", "--cap-drop=all", "--security-opt=no-new-privileges", "--memory=512m", "--cpus=2", "--pids-limit=128",
168 "--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=64m", "--publish=127.0.0.1:" + str(backend_port) + ":8000",
169 "--add-host=keycloak.studio.test:host-gateway", "--volume=/var/lib/studio/ca-bundle.crt:/etc/ssl/certs/ca-certificates.crt:ro",
170 "--env=DOMAIN=" + host + ":" + str(port), "--env=SERVER_TITLE=MCP fixture",
171 "--env=SESSION_SECRET=" + uuid.uuid4().hex + uuid.uuid4().hex,
172 "--env=OAUTH2_CLIENT=oidc,keycloak.studio.test/realms/master|" + client_name + "|" + client_secret,
173 "--env=NPROC=2", "--env=MIRRORS_DISABLE_CLONE=true", "--env=MIRRORS_DISABLE_PULL_POLL=true",
174 *(["--env=OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=" + traces_endpoint] if traces_env else []),
175 *volumes, image, *(["--otel-exporter-otlp-traces-endpoint", traces_endpoint] if traces_endpoint and not traces_env else []))
176 deployed = json.loads(run("podman", "inspect", container))[0]
177 assert deployed["ImageName"] == image
178 assert deployed["Config"]["User"] == "65534:65534"
179 context = ssl.create_default_context(cafile="/var/lib/studio/ca-bundle.crt")
180 from api import LoopbackHTTPS
181
182 class HTTPS(urllib.request.HTTPSHandler):
183 def https_open(self, request):
184 assert urllib.parse.urlsplit(request.full_url).netloc == urllib.parse.urlsplit(origin).netloc
185 return self.do_open(LoopbackHTTPS, request, context=context)
186
187 readiness = urllib.request.build_opener(HTTPS())
188 deadline = time.monotonic() + 15
189 while True:
190 try:
191 with readiness.open(origin, timeout=2) as response:
192 assert response.status == 200
193 break
194 except OSError:
195 if time.monotonic() >= deadline:
196 raise AssertionError((root / "caddy.log").read_text())
197 time.sleep(.1)
198 yield origin, root / "data/astheno.shale.db", container
199 except Exception:
200 if subprocess.run(["podman", "container", "exists", container], capture_output=True).returncode == 0:
201 state = json.loads(run("podman", "inspect", container))[0]["State"]
202 print(json.dumps({"fixture_backend_running": state["Running"], "fixture_backend_exit_code": state["ExitCode"],
203 "fixture_backend_oom_killed": state["OOMKilled"]}), flush=True)
204 raise
205 finally:
206 try:
207 subprocess.run(["podman", "rm", "--ignore", "--force", container], check=True, capture_output=True)
208 finally:
209 try:
210 if caddy:
211 caddy.terminate()
212 try:
213 caddy.wait(timeout=10)
214 except subprocess.TimeoutExpired:
215 caddy.kill()
216 caddy.wait(timeout=10)
217 finally:
218 if proxy:
219 proxy.shutdown()
220 proxy.server_close()
221 proxy_thread.join(timeout=5)
222 from api import Keycloak
223 keycloak = Keycloak(keycloak.host, importlib.import_module("dashboard-run").secret("get", "keycloak", "password"), attempts=1)
224 if client_id is None:
225 clients = keycloak.request("/admin/realms/master/clients?clientId=" + client_name)
226 assert len(clients) <= 1
227 client_id = clients[0]["id"] if clients else None
228 if client_id:
229 keycloak.request("/admin/realms/master/clients/" + client_id, "DELETE")
230
231
232
233def main():
234 parser = argparse.ArgumentParser()
235 parser.add_argument("--output", type=Path)
236 parser.add_argument("--image")
237 parser.add_argument("--traces-endpoint")
238 parser.add_argument("--traces-env", action="store_true")
239 args = parser.parse_args()
240 if args.output:
241 args.output.unlink(missing_ok=True)
242 assert not args.traces_env or args.traces_endpoint
243 if args.traces_endpoint:
244 endpoint = urllib.parse.urlsplit(args.traces_endpoint)
245 assert endpoint.scheme in {"http", "https"} and endpoint.hostname
246 assert not endpoint.username and not endpoint.password
247 repo = Path(__file__).resolve().parent.parent
248 image = args.image or re.search(r'^\s*image = "([^"]+)"', (repo / "service/shale/service.pkl").read_text(), re.MULTILINE).group(1)
249 sys.path.insert(0, str(repo / "tools"))
250 sys.path.insert(0, str(repo / "service/keycloak"))
251 from api import Keycloak, LoopbackHTTPS
252 keycloak_host = "keycloak.studio.test"
253 shale_host = "localhost"
254 context = ssl.create_default_context(cafile="/var/lib/studio/ca-bundle.crt")
255
256 class HTTPS(urllib.request.HTTPSHandler, urllib.request.HTTPHandler):
257 def http_open(self, request):
258 raise AssertionError("unencrypted fixture redirect")
259
260 def https_open(self, request):
261 parts = urllib.parse.urlsplit(request.full_url)
262 assert parts.netloc in {keycloak_host, urllib.parse.urlsplit(shale).netloc}
263 return self.do_open(LoopbackHTTPS, request, context=context)
264
265 class NoRedirect(urllib.request.HTTPRedirectHandler):
266 def redirect_request(self, req, fp, code, msg, headers, newurl):
267 return None
268
269 key = uuid.uuid4().hex
270 name = "mcp-fixture-" + key
271 password = uuid.uuid4().hex + "A1!"
272 keycloak = Keycloak(keycloak_host, importlib.import_module("dashboard-run").secret("get", "keycloak", "password"))
273 cookies = http.cookiejar.CookieJar()
274 opener = urllib.request.build_opener(HTTPS(), urllib.request.HTTPCookieProcessor(cookies))
275
276 def request(url, fields=None, headers=None, client=opener):
277 headers = dict(headers or {})
278 data = None
279 if fields is not None:
280 parts = urllib.parse.urlsplit(url)
281 headers.update(Origin=parts.scheme + "://" + parts.netloc, Referer=parts._replace(fragment="").geturl())
282 data = urllib.parse.urlencode(fields).encode()
283 req = urllib.request.Request(url, headers=headers, data=data)
284 try:
285 response = client.open(req, timeout=15)
286 except urllib.error.HTTPError as error:
287 response = error
288 with response:
289 return response.status, response.url, response.headers, response.read().decode()
290
291 def submit(url, form, changes, client=opener):
292 target = urllib.parse.urljoin(url, form.get("action", ""))
293 assert urllib.parse.urlsplit(target).netloc == urllib.parse.urlsplit(shale).netloc
294 fields = {a["name"]: a.get("value", "") for a in form["fields"] if a.get("name") and a.get("type") == "hidden"}
295 fields.update(timezone="UTC", tzoffset="+00:00", **changes)
296 return request(target, fields, client=client)
297
298 accounts = [(name, password), ("mcp-stranger-" + key, uuid.uuid4().hex + "A1!")]
299 try:
300 for username, credential in accounts:
301 keycloak.request("/admin/realms/master/users", "POST", {
302 "username": username, "firstName": username, "lastName": "Fixture", "email": username + "@fixture.invalid",
303 "emailVerified": True, "enabled": True,
304 "credentials": [{"type": "password", "value": credential, "temporary": False}],
305 })
306 users = keycloak.request("/admin/realms/master/users?username=" + username + "&exact=true")
307 assert len(users) == 1
308 with instance(keycloak, key, image, args.traces_endpoint, args.traces_env) as (shale, database, container):
309 def login(client, username, credential):
310 status, url, _, body = request(shale + "/-/login", client=client)
311 assert status == 200, status
312 form = next(f for f in Page(body).forms if any(a.get("name") == "password" for a in f["fields"]))
313 fields = {a["name"]: a.get("value", "") for a in form["fields"] if a.get("name")}
314 fields.update(username=username, password=credential)
315 status, url, _, _ = request(urllib.parse.urljoin(url, form["action"]), fields, client=client)
316 assert status == 200 and urllib.parse.urlsplit(url).hostname == shale_host
317
318 login(opener, *accounts[0])
319 stranger = urllib.request.build_opener(HTTPS(), urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
320 login(stranger, *accounts[1])
321 status, url, _, body = request(shale + "/-/settings")
322 assert status == 200, status
323 form = next(f for f in Page(body).forms if any(a.get("name") == "label" for a in f["fields"]))
324 assert form.get("enctype", "application/x-www-form-urlencoded") == "application/x-www-form-urlencoded"
325 status, _, _, body = submit(url, form, {"label": name})
326 if status != 200:
327 state = json.loads(subprocess.check_output(["podman", "inspect", container], text=True))[0]["State"]
328 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
329 count = db.execute("SELECT count(*) FROM personal_tokens").fetchone()[0]
330 result = {"oidc_fixture_user": True, "mint_status": status,
331 "backend_exit_code": state["ExitCode"], "token_rows_after_response": count}
332 print(json.dumps(result), flush=True)
333 if args.output:
334 args.output.write_text(json.dumps(result, indent=2) + "\n")
335 raise AssertionError("token mint failed; write outcome recorded without replay")
336 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
337 users = db.execute("SELECT id FROM users WHERE name=?", (name,)).fetchall()
338 assert len(users) == 1
339 identity = users[0][0]
340 tokens = db.execute("SELECT uuid FROM personal_tokens WHERE agent=? AND label=?", (identity, name)).fetchall()
341 assert len(tokens) == 1
342 token_id = tokens[0][0]
343 page = Page(body)
344 tokens = {text.strip() for text in [*page.codes, *[a.get("value", "") for f in page.forms for a in f["fields"]]]
345 if re.fullmatch(r"[A-Za-z0-9_:.=+-]{24,}", text.strip())}
346 assert len(tokens) == 1
347 token = next(iter(tokens))
348 anonymous = urllib.request.build_opener(HTTPS(), NoRedirect())
349 for authorization in ["Bearer " + token, "token " + token,
350 "Basic " + base64.b64encode((name + ":" + token).encode()).decode()]:
351 status, _, _, body = request(shale + "/-/settings", headers={"Authorization": authorization}, client=anonymous)
352 assert status == 303 and name not in body
353 status, url, _, body = request(shale + "/-/new")
354 assert status == 200
355 form = next(f for f in Page(body).forms if any(a.get("name") == "name" for a in f["fields"]))
356 status, url, _, body = submit(url, form, {"name": "fixture", "description": "Disposable MCP adapter test", "access": "private"})
357 assert status == 200, status
358 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
359 assert db.execute("SELECT count(*) FROM repositories WHERE name='fixture' AND owner=?", (identity,)).fetchone()[0] == 1
360 repository_url = url
361 for client in (anonymous, stranger):
362 status, _, _, _ = request(repository_url, client=client)
363 assert status in (403, 404), status
364 issues = next(link for link in Page(body).links if "issues" in link)
365 status, url, _, body = request(urllib.parse.urljoin(repository_url, issues))
366 assert status == 200
367 new_issue = next(link for link in Page(body).links if link == "./new")
368 status, url, _, body = request(urllib.parse.urljoin(url, new_issue))
369 assert status == 200
370 title = "Fixture <angle> & Unicode ☃"
371 description = "Disposable body <script>fixture</script> & Unicode ☃"
372 form = next(f for f in Page(body).forms if any(a.get("name") == "title" for a in f["fields"]))
373 status, url, _, body = submit(url, form, {"title": title, "description": description})
374 assert status == 200, status
375 issue_url = url
376 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
377 issues = db.execute("SELECT id,title FROM issues").fetchall()
378 assert len(issues) == 1 and issues[0][1] == title
379 assert db.execute("SELECT payload FROM issue_actions WHERE issue=?", (issues[0][0],)).fetchone()[0] == description
380 status, _, _, _ = request(issue_url, client=anonymous)
381 assert status in (403, 404), status
382 status, _, _, body = request(issue_url)
383 assert status == 200
384 page = Page(body)
385 comment = "Fixture comment <angle> & Unicode ☃"
386 form = next(f for f in page.forms if any(a.get("name") == "t" and a.get("value") == "comment" for a in f["fields"]))
387 status, _, _, body = submit(issue_url, form, {"comment": comment})
388 assert status == 200, status
389 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
390 assert db.execute("SELECT count(*) FROM issue_actions WHERE payload=?", (comment,)).fetchone()[0] == 1
391 form = next(f for f in Page(body).forms if any(a.get("name") == "t" and a.get("value") == "status" for a in f["fields"]))
392 status_field = next(a for a in form["fields"] if a.get("name") == "status")
393 assert any(option["value"] == "done" for option in status_field["options"])
394 status, _, _, body = submit(issue_url, form, {"status": "done"})
395 assert status == 200
396 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
397 assert db.execute("SELECT status FROM issues").fetchone()[0] == "done"
398 status, _, _, stranger_settings = request(shale + "/-/settings", client=stranger)
399 assert status == 200
400 stranger_csrf = {a["name"]: a["value"] for f in Page(stranger_settings).forms for a in f["fields"] if a.get("name") == "csrf_token"}
401 for client, csrf in [(anonymous, {}), (stranger, stranger_csrf)]:
402 status, _, _, _ = request(issue_url, {"t": "comment", "comment": "UNAUTHORIZED",
403 "timezone": "UTC", "tzoffset": "+00:00", **csrf}, client=client)
404 assert status in ((400, 403, 404) if client is anonymous else (403, 404)), status
405 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
406 assert db.execute("SELECT count(*) FROM issue_actions WHERE payload='UNAUTHORIZED'").fetchone()[0] == 0
407 subprocess.run(["podman", "restart", container], check=True, capture_output=True)
408 deadline = time.monotonic() + 15
409 while True:
410 status, _, _, body = request(issue_url)
411 if status == 200:
412 break
413 assert time.monotonic() < deadline, status
414 time.sleep(.1)
415 assert "Unicode ☃" in body
416 for client in (anonymous, stranger):
417 status, _, _, _ = request(issue_url, client=client)
418 assert status in (403, 404), status
419 status, url, _, body = request(shale + "/-/settings")
420 form = next(f for f in Page(body).forms if any(a.get("name") == "label" for a in f["fields"]))
421 status, _, _, body = submit(url, form, {"label": "second"})
422 assert status == 200, status
423 form = next(f for f in Page(body).forms if any(a.get("name") == "t" and a.get("value") == "token_revoke" for a in f["fields"]))
424 status, _, _, _ = submit(url, form, {})
425 assert status == 200
426 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
427 assert db.execute("SELECT count(*) FROM personal_tokens WHERE agent=?", (identity,)).fetchone()[0] == 1
428 assert db.execute("SELECT count(*) FROM personal_tokens WHERE uuid=?", (token_id,)).fetchone()[0] == 0
429 captured = http.cookiejar.CookieJar()
430 for cookie in cookies:
431 if cookie.domain.startswith("localhost"):
432 captured.set_cookie(copy.copy(cookie))
433 assert len(captured) > 0
434 replay = urllib.request.build_opener(HTTPS(), NoRedirect(), urllib.request.HTTPCookieProcessor(captured))
435 status, _, _, _ = request(issue_url, client=replay)
436 assert status == 200
437 status, _, _, _ = request(shale + "/-/logout")
438 assert status == 200
439 status, _, _, _ = request(issue_url)
440 assert status in (403, 404), status
441 replay_status, _, _, _ = request(issue_url, client=replay)
442 assert replay_status in (200, 303, 403, 404), replay_status
443 with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as db:
444 session_rows = db.execute("SELECT count(*) FROM sessions WHERE user=?", (identity,)).fetchone()[0]
445 result = {"backend_image": image,
446 "backend_image_id": json.loads(subprocess.check_output(["podman", "inspect", container], text=True))[0]["Image"],
447 "traces_endpoint": args.traces_endpoint, "traces_environment_variable": args.traces_env,
448 "logout_cookie_replay_status": replay_status,
449 "backend_session_rows_after_logout": session_rows,
450 "checks": ["two_user_oidc", "personal_token_mint", "pat_not_html_auth", "private_repository_create",
451 "anonymous_read_denied", "other_user_read_denied", "issue_create_unicode", "comment_create_unicode",
452 "issue_close", "anonymous_write_denied", "other_user_write_denied", "session_survives_restart",
453 "second_token_mint", "token_revoke", "logout_denies_private_read",
454 "cross_user_denied_after_restart", "captured_session_valid_before_logout", "logout_cookie_replay_probe"]}
455 finally:
456 for username, _ in accounts:
457 users = keycloak.request("/admin/realms/master/users?username=" + username + "&exact=true")
458 for user in users:
459 assert user["username"] == username
460 keycloak.request("/admin/realms/master/users/" + user["id"], "DELETE")
461 result["checks"].append("fixture_cleanup")
462 print(json.dumps(result), flush=True)
463 if args.output:
464 args.output.write_text(json.dumps(result, indent=2) + "\n")
465
466
467if __name__ == "__main__":
468 main()