Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289"""Web lifecycle, sign-in, and cross-release HTML caching through the real CLI/daemon."""
import http.cookiejarimport jsonimport osimport reimport socketimport subprocessimport sysimport tempfileimport unittestimport urllib.errorimport urllib.requestfrom pathlib import Path
from harness import Albedo, exclusive_class
# A tailscale CLI on PATH that keeps its serve config in $FAKE_TAILSCALE, so# --tailnet-https runs end to end without touching a real tailnet.FAKE_TAILSCALE = """import json, os, sysroot = os.environ["FAKE_TAILSCALE"]path = os.path.join(root, "state.json")with open(path) as file: state = json.load(file)args = sys.argv[1:]lost = Falsewith open(os.path.join(root, "calls"), "a") as file: file.write(" ".join(args) + "\\n")if args[:2] == ["status", "--json"]: print(json.dumps({"BackendState": "Running", "Self": {"DNSName": "albedo.example.ts.net."}, "CertDomains": state["certs"]}))elif args == ["serve", "status", "--json"]: lost = state.pop("lose_status", False) if not lost: print(json.dumps(state["serve"]))elif args[:3] == ["serve", "--bg", "--https=443"]: root = {"Handlers": {"/": {"Proxy": args[3]}}} state["serve"] = {"TCP": {"443": {"HTTPS": True}}, "Web": {"albedo.example.ts.net:443": root}} state["lose_status"] = state.pop("flaky", False)elif args == ["serve", "--https=443", "--set-path", "/", "off"]: state["serve"] = {}else: sys.exit("unexpected tailscale " + " ".join(args))with open(path, "w") as file: json.dump(state, file)if lost: sys.exit("tailscaled went away")"""
@exclusive_classclass WebTest(unittest.TestCase): def setUp(self): self.app = Albedo(providers={}).__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.addCleanup(self.app.cli, "web", "--stop") (self.app.workspace / "index.html").write_text("<p>web fixture</p>") self.app.env["ALBEDO_WEB_DIST"] = str(self.app.workspace)
def web(self, *args): out = self.app.cli("web", "--local", "--no-open", *args) match = re.search(r"http://127\.0\.0\.1:\d+/\?key=\S+", out) self.assertIsNotNone(match, out) assert match is not None return match[0]
def tailnet(self, serve=None, certs=("albedo.example.ts.net",), **flags): root = Path(tempfile.mkdtemp(dir=self.app.root)) (root / "state.json").write_text( json.dumps({"serve": serve or {}, "certs": certs, **flags}) ) script = root / "tailscale" script.write_text(f"#!{sys.executable}\n{FAKE_TAILSCALE}") script.chmod(0o755) self.app.env["FAKE_TAILSCALE"] = str(root) self.app.env["PATH"] = f"{root}{os.pathsep}{os.environ['PATH']}" return root
def refused(self, *args): command = self.app.daemon.start_cli( "web", "--no-open", *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) out, err = command.communicate(timeout=60) self.assertNotEqual(command.returncode, 0, out) return err.decode()
def browser(self, link): jar = http.cookiejar.CookieJar() client = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) with client.open(link, timeout=3) as response: self.assertIn(b"web fixture", response.read()) # A browser closing must not sign it out: the cookie has an expiry. (cookie,) = list(jar) self.assertIsNotNone(cookie.expires) return client
def test_background_lifecycle_and_sign_in(self): link = self.web("--port", "0") client = self.browser(link) base = link.split("/?")[0] with client.open(base + "/api/server", timeout=3) as response: self.assertEqual(json.load(response)["state"], "ready") self.assertEqual(self.web(), link) rotated = self.web("--new-link") self.assertNotEqual(rotated, link) with self.assertRaises(urllib.error.HTTPError) as caught: client.open(rotated.split("/?")[0] + "/api/sessions", timeout=3) self.assertEqual(caught.exception.code, 401) client = self.browser(rotated) link = rotated base = rotated.split("/?")[0] self.app.cli("web", "--stop") with self.assertRaises(urllib.error.URLError): client.open(base + "/api/sessions", timeout=3) restarted = self.web("--port", "0") self.assertEqual(restarted.split("key=")[1], link.split("key=")[1])
def test_stop_stops_a_server_too_busy_to_answer(self): # Orphaned, so init reaps it and --stop sees it go. pid = int( subprocess.check_output(["sh", "-c", "sleep 60 >/dev/null & echo $!"]) ) self.addCleanup( lambda: subprocess.run(["kill", "-9", str(pid)], capture_output=True) ) with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) listener.listen() port = listener.getsockname()[1] (self.app.home / "web.json").write_text( json.dumps( {"pid": pid, "port": port, "url": f"http://127.0.0.1:{port}"} ) ) self.app.cli("web", "--stop") with self.assertRaises(ProcessLookupError, msg="--stop left a busy server"): os.kill(pid, 0)
def test_html_revalidation_tracks_content_not_packaged_mtime(self): index = self.app.workspace / "index.html" first = b"<!doctype html><p>web fixture one</p>" second = b"<!doctype html><p>web fixture two</p>" legacy_date = "Thu, 01 Jan 1970 00:00:01 GMT" index.write_bytes(first) os.utime(index, (1, 1)) assets = self.app.workspace / "assets" assets.mkdir() asset = assets / "index-abc.js" asset.write_text("console.log('fixture')") os.utime(asset, (1, 1)) link = self.web("--port", "0") client = self.browser(link) base = link.split("/?")[0]
def fetch(path="/", headers=None, method="GET"): request = urllib.request.Request( base + path, headers=headers or {}, method=method ) try: response = client.open(request, timeout=3) except urllib.error.HTTPError as error: response = error with response: return response.status, response.headers, response.read()
status, before, body = fetch() self.assertEqual((status, body), (200, first)) # Nix gives different builds the same timestamp and these pages also # have equal sizes. Neither is a valid cross-release HTML validator. index.write_bytes(second) os.utime(index, (1, 1)) status, after, body = fetch(headers={"If-Modified-Since": legacy_date}) self.assertEqual((status, body), (200, second)) self.assertEqual(after["Cache-Control"], "no-cache") self.assertNotIn("Last-Modified", after) self.assertIsNotNone(before["ETag"]) self.assertIsNotNone(after["ETag"]) self.assertNotEqual(before["ETag"], after["ETag"])
for method, expected in [("GET", second), ("HEAD", b"")]: with self.subTest(stale_etag=method): status, headers, body = fetch( headers={ "If-None-Match": before["ETag"], "If-Modified-Since": legacy_date, }, method=method, ) self.assertEqual((status, body), (200, expected)) self.assertEqual(headers["ETag"], after["ETag"]) self.assertEqual(int(headers["Content-Length"]), len(second))
for etag in [after["ETag"], "W/" + after["ETag"]]: with self.subTest(current_etag=etag): status, headers, body = fetch( headers={"If-None-Match": etag, "If-Modified-Since": legacy_date} ) self.assertEqual((status, body), (304, b"")) self.assertEqual(headers["Cache-Control"], "no-cache")
status, headers, body = fetch("/index.html") self.assertEqual((status, body), (200, second)) self.assertEqual(headers["ETag"], after["ETag"]) status, headers, _ = fetch("/assets/index-abc.js") self.assertEqual(status, 200) self.assertEqual( headers["Cache-Control"], "public, max-age=31536000, immutable" ) status, _, body = fetch( "/assets/index-abc.js", {"If-Modified-Since": legacy_date} ) self.assertEqual((status, body), (304, b"")) status, _, body = fetch("/api/server") self.assertEqual(status, 200) self.assertEqual(json.loads(body)["state"], "ready") index.unlink() status, _, _ = fetch() self.assertEqual(status, 404)
def test_busy_port_explains_startup_failure_without_waiting(self): with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) listener.listen() command = self.app.daemon.start_cli( "web", "--local", "--no-open", "--port", str(listener.getsockname()[1]), stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) try: _, err = command.communicate(timeout=10) finally: if command.poll() is None: command.kill() command.communicate() self.assertNotEqual(command.returncode, 0) self.assertIn("address already in use", err.decode().lower())
def test_tailnet_https_publishes_and_stop_removes_it(self): root = self.tailnet() out = self.app.cli("web", "--tailnet-https", "--no-open", "--port", "0") self.assertRegex(out, r"serving at https://albedo\.example\.ts\.net/\?key=") self.assertNotIn("http://albedo.example.ts.net", out) proxy = json.loads((root / "state.json").read_text())["serve"]["Web"][ "albedo.example.ts.net:443" ]["Handlers"]["/"]["Proxy"] port = json.loads((self.app.home / "web.json").read_text())["port"] self.assertEqual(proxy, f"http://127.0.0.1:{port}") self.app.cli("web", "--stop") self.assertEqual(json.loads((root / "state.json").read_text())["serve"], {})
def test_tailnet_https_refuses_without_changing_tailscale(self): with self.subTest("no certificate, as on Headscale"): root = self.tailnet(certs=None) self.assertIn( "issues no certificate", self.refused("--tailnet-https", "--port", "0") ) self.assertNotIn("serve", (root / "calls").read_text()) with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] with self.subTest("our root opened to the internet by Funnel"): funnel = { "Web": { "albedo.example.ts.net:443": { "Handlers": {"/": {"Proxy": f"http://127.0.0.1:{port}"}} } }, "AllowFunnel": {"albedo.example.ts.net:443": True}, } root = self.tailnet(serve=funnel) self.assertIn( "AllowFunnel", self.refused("--tailnet-https", "--port", str(port)) ) self.assertEqual( json.loads((root / "state.json").read_text())["serve"], funnel ) with self.subTest("a handler created but not confirmed"): root = self.tailnet(flaky=True) self.assertIn( "tailscaled went away", self.refused("--tailnet-https", "--port", str(port)), ) self.assertEqual(json.loads((root / "state.json").read_text())["serve"], {}) self.assertFalse((self.app.home / "web.json").exists())