#!/usr/bin/env python3 """Real upstream HTTPS/reverse-proxy tests; optionally drive the Android emulator. Keys and CA trust exist only in a private temporary directory and the .integration APK. Production and ordinary debug builds never include this trust configuration. """ import argparse import http.cookiejar import json import os from pathlib import Path import secrets import shutil import ssl import subprocess import tempfile import time import urllib.error import urllib.request ROOT = Path(__file__).resolve().parents[1] REVISION = "10bfef7686ed7303b4c74c7fcc55a5eb16356c9d" SCREENSHOTS = ("native-dashboard.png",) + tuple( f"{page}-{theme}.png" for page in ("login", "dashboard", "tasks", "notes", "shopping") for theme in ("light", "dark")) VISUAL_OUTPUTS = SCREENSHOTS + ("viewport.json",) def prepare_android_screenshots(app_output): additional = app_output / "connected_android_test_additional_output/integrationAndroidTest" screenshots = app_output / "native-screenshots" # Never accept a previous run's screenshot as current evidence. for name in VISUAL_OUTPUTS: for stale in additional.rglob(name): stale.unlink() (screenshots / name).unlink(missing_ok=True) return additional, screenshots def collect_native_screenshot(additional, screenshots, gallery=False): names = VISUAL_OUTPUTS if gallery else ("native-dashboard.png",) validated = {} parents = set() for name in names: captures = list(additional.rglob(name)) assert len(captures) == 1, f"Missing or ambiguous native fixture screenshot: {name}" capture = captures[0].read_bytes() if name.endswith(".png"): assert capture.startswith(b"\x89PNG\r\n\x1a\n"), "Invalid native fixture screenshot" else: viewport = json.loads(capture) assert viewport["widthPx"] > 0 and viewport["heightPx"] > 0 and viewport["density"] > 0 parents.add(captures[0].parent) validated[name] = capture assert len(parents) == 1, "Mixed-device native gallery evidence" screenshots.mkdir(parents=True, exist_ok=True) for name, capture in validated.items(): (screenshots / name).write_bytes(capture) def run(upstream, android, cookie_without_secure=False): revision = subprocess.check_output(["git", "-C", str(upstream), "rev-parse", "HEAD"], text=True).strip() assert revision == REVISION, f"Expected pinned upstream {REVISION}, got {revision}" base_temp = Path("/tmp/opencode") if Path("/tmp/opencode").exists() else Path(tempfile.gettempdir()) with tempfile.TemporaryDirectory(prefix="yuvomi-https-", dir=base_temp) as temporary: directory = Path(temporary) subprocess.run([ "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", "-keyout", str(directory / "key.pem"), "-out", str(directory / "cert.pem"), "-subj", "/CN=Yuvomi temporary test CA", "-addext", "subjectAltName=IP:127.0.0.1,IP:10.0.2.2,DNS:localhost", ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) with (directory / "server.log").open("w") as log: process = subprocess.Popen(["node", str(ROOT / "scripts/integration-server.mjs"), str(upstream), temporary], cwd=upstream, stdout=log, stderr=log, env={**os.environ, "NODE_EXTRA_CA_CERTS": str(directory / "cert.pem"), "YUVOMI_TEST_SESSION_SECURE": "false" if cookie_without_secure else "true"}) try: for _ in range(200): if (directory / "ready.json").exists(): break if process.poll() is not None: raise RuntimeError((directory / "server.log").read_text()) time.sleep(0.1) ready = json.loads((directory / "ready.json").read_text()) port = ready["port"] base = f"https://127.0.0.1:{port}" context = ssl.create_default_context(cafile=str(directory / "cert.pem")) cookies = http.cookiejar.CookieJar() client = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context), urllib.request.HTTPCookieProcessor(cookies)) def request(route, payload=None, headers=None): body = None if payload is None else json.dumps(payload).encode() req = urllib.request.Request(base + route, body, headers={ "Accept": "application/json", "Content-Type": "application/json", **(headers or {})}) try: with client.open(req, timeout=15) as response: return response.status, response.read(), response.headers except urllib.error.HTTPError as error: return error.code, error.read(), error.headers # A certificate unknown to the system MUST be rejected. try: urllib.request.urlopen(base + "/api/v1/version", timeout=5) raise AssertionError("Untrusted certificate was accepted") except urllib.error.URLError as error: assert isinstance(error.reason, ssl.SSLCertVerificationError) print("PASS: untrusted TLS certificate rejected") status, body, _ = request("/api/v1/version") metadata = json.loads(body) assert status == 200 and isinstance(metadata["app_name"], str) assert metadata["setup_required"] is True and "version" not in metadata assert isinstance(metadata["password_reset_enabled"], bool) status, body, _ = request("/api/v1/auth/oidc/config") assert status == 200 and json.loads(body)["enabled"] is True print("PASS: actual upstream compatibility metadata over HTTPS reverse proxy") password = secrets.token_urlsafe(24) assert request("/api/v1/auth/setup", {"username": "mobiletest", "display_name": "Mobile Test", "password": password})[0] == 201 status, body, headers = request("/api/v1/auth/login", {"username": "mobiletest", "password": password}) assert status == 200 session = [cookie for cookie in cookies if cookie.name == "yuvomi.sid"][0] assert session.secure == (not cookie_without_secure) and session.has_nonstandard_attr("HttpOnly") assert "SameSite=Lax" in " ".join(headers.get_all("Set-Cookie")) csrf = json.loads(body)["csrfToken"] assert request("/api/v1/auth/me")[0] == 200 assert json.loads(request("/api/v1/version")[1])["version"] == "2.78.0" assert request("/api/v1/auth/logout", {})[0] == 403 assert request("/api/v1/auth/logout", {}, {"X-CSRF-Token": csrf})[0] == 200 assert request("/api/v1/auth/me")[0] == 401 cookie_mode = "HTTPS with upstream-default non-Secure session cookie" if cookie_without_secure else "Secure session cookie" print(f"PASS: login, {cookie_mode}, HttpOnly/SameSite, persistence, CSRF rejection and logout/revocation") assert request("/api/v1/auth/oidc/start")[0] == 200 status, body, _ = request("/api/v1/auth/me") identity = json.loads(body) assert status == 200 and identity.get("user", {}).get("username") == "mobile-oidc", ( f"OIDC session rejected (HTTP {status}); " + "\n".join( (directory / "server.log").read_text().splitlines()[-20:])) assert request("/api/v1/auth/logout", {}, {"X-CSRF-Token": identity["csrfToken"]})[0] == 200 print("PASS: upstream OIDC discovery, RS256/nonce validation, S256 PKCE exchange and test-IdP session") if android: resources = directory / "res" (resources / "xml").mkdir(parents=True) (resources / "raw").mkdir() shutil.copyfile(directory / "cert.pem", resources / "raw/integration_ca.pem") (resources / "xml/network_security_config.xml").write_text(''' ''') env = {**os.environ, "YUVOMI_TEST_RES": str(resources), "YUVOMI_TEST_URL": base, "YUVOMI_TEST_PASSWORD": password} sdk = Path(env.get("ANDROID_HOME", env.get("ANDROID_SDK_ROOT", ""))) adb = sdk / "platform-tools/adb" user_state = subprocess.check_output( [str(adb), "shell", "am", "get-started-user-state", "0"], text=True).strip() if user_state != "RUNNING_UNLOCKED": raise RuntimeError(f"Unlock the emulator before testing (user state: {user_state}).") subprocess.run([str(adb), "reverse", f"tcp:{port}", f"tcp:{port}"], check=True) subprocess.run([str(adb), "reverse", f"tcp:{ready['idpPort']}", f"tcp:{ready['idpPort']}"], check=True) try: app_output = (Path(env["YUVOMI_BUILD_DIR"]) / "app" if "YUVOMI_BUILD_DIR" in env else ROOT / "android/app/build") / "outputs" additional, screenshots = prepare_android_screenshots(app_output) # The ephemeral password is passed through the local test runner only, # never written into a source file or printed by this harness. result = subprocess.run(["bash", str(ROOT / "android/gradlew"), "-p", str(ROOT / "android"), "-Pintegration", "connectedIntegrationAndroidTest"], env=env, check=False) if result.returncode: raise RuntimeError("Android integration tests failed. Inspect the Gradle test report.") reports = app_output / "androidTest-results/connected/integration" subprocess.run(["python3", str(ROOT / "scripts/verify-test-results.py"), str(reports), "--minimum", "10"], check=True) collect_native_screenshot(additional, screenshots, gallery=True) finally: subprocess.run([str(adb), "reverse", "--remove", f"tcp:{port}"], check=False) subprocess.run([str(adb), "reverse", "--remove", f"tcp:{ready['idpPort']}"], check=False) finally: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait() if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("upstream", type=Path) parser.add_argument("--android", action="store_true") parser.add_argument("--cookie-without-secure", action="store_true", help="Test upstream's default session-cookie attributes over HTTPS; never enables HTTP") arguments = parser.parse_args() run(arguments.upstream.resolve(), arguments.android, arguments.cookie_without_secure)