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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187"""Detached web lifecycle and browser sign-in through the real CLI and daemon."""
import http.cookiejarimport jsonimport osimport reimport socketimport subprocessimport sysimport tempfileimport unittestimport urllib.errorimport urllib.requestfrom pathlib import Path
from harness import Albedo, exclusive
# 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")"""
@exclusiveclass 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_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())