SHA256
Yuvomi Mobile Contributors
9ca11e4fb5
Add an ephemeral HTTPS proxy and standards-shaped RS256/S256-PKCE test IdP around the unmodified upstream checkout. Require unlocked devices and nonempty, failure-free JUnit evidence. The extended API 35 suite passes all seven cases, including native Blob/file transfer, session recreation, outage recovery, isolated OIDC and back navigation; eight evidence-verifier regressions also pass. API 37 and real-provider/device validation remain release gates.
153 lines
9.3 KiB
Python
153 lines
9.3 KiB
Python
#!/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"
|
|
|
|
|
|
def run(upstream, android):
|
|
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")})
|
|
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 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
|
|
print("PASS: login, Secure/HttpOnly/SameSite cookies, 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('''<network-security-config>
|
|
<base-config cleartextTrafficPermitted="false"><trust-anchors>
|
|
<certificates src="system"/><certificates src="@raw/integration_ca"/>
|
|
</trust-anchors></base-config></network-security-config>''')
|
|
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:
|
|
# 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.")
|
|
output = Path(env.get("YUVOMI_BUILD_DIR", ROOT / "android"))
|
|
reports = output / "app/outputs/androidTest-results/connected/integration"
|
|
if "YUVOMI_BUILD_DIR" not in env:
|
|
reports = ROOT / "android/app/build/outputs/androidTest-results/connected/integration"
|
|
subprocess.run(["python3", str(ROOT / "scripts/verify-test-results.py"),
|
|
str(reports), "--minimum", "7"], check=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")
|
|
arguments = parser.parse_args()
|
|
run(arguments.upstream.resolve(), arguments.android)
|