SHA256
test: exercise native Android flows against pinned upstream HTTPS and OIDC
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.
This commit is contained in:
1 parent
f7b9dcd91e
commit
9ca11e4fb5
7 files changed
+691
No files matched your search
@@ -0,0 +1,152 @@
|
||||
#!/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)
|
||||
Reference in new issue
Block a user