tegwick
f6eec13f7c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1abf1-7e95-7141-87c9-ee78332bf790
83 lines
4.2 KiB
Python
83 lines
4.2 KiB
Python
"""Manual native lifecycle fixture: disposable local page, no identity request.
|
|
|
|
The production worker's issuer/callback guard is first checked to reject the
|
|
local URL. A separate in-memory fixture copy admits only the exact disposable
|
|
loopback listener. Production source is never modified. No login, real profile
|
|
or token helper is used. Run only with explicit GUI approval.
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--module", type=Path, default=Path(__file__).resolve().parents[2]
|
|
/ "railiance-platform/scripts/operator-browser/windows_browser.py")
|
|
args = parser.parse_args()
|
|
spec = importlib.util.spec_from_file_location("native_lifecycle_fixture", args.module)
|
|
native = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(native)
|
|
visits = []
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
visits.append(self.path)
|
|
if self.path == "/":
|
|
self.send_response(302)
|
|
self.send_header("Location", "http://localhost:" + str(self.server.server_port) + "/ready")
|
|
self.end_headers()
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html")
|
|
self.end_headers()
|
|
self.wfile.write(b"<h1>Native browser lifecycle check</h1><p>No login or credentials. This window closes automatically.</p>")
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
url = "http://127.0.0.1:" + str(server.server_port) + "/"
|
|
payload = json.dumps({"url": url}).encode()
|
|
try:
|
|
refused = subprocess.run(native.command(10), input=payload, capture_output=True, timeout=20, check=False)
|
|
assert refused.stdout.strip() == b"failed"
|
|
source = native.SCRIPT.replace("__MAXIMUM__", "10")
|
|
original = "if($uri.Scheme -ne 'https' -or $uri.Host -ne 'kc.coulomb.social' -or $uri.Port -ne 443 -or $uri.UserInfo -or $uri.Fragment){throw 'invalid origin'}"
|
|
assert source.count(original) == 1
|
|
source = source.replace(original, "if($url -ne '" + url + "'){throw 'fixture origin mismatch'}")
|
|
original = "if($callbacks.Count -ne 1 -or $callbacks[0] -notin @('http://localhost:8250/oidc/callback','http://127.0.0.1:8250/oidc/callback')){throw 'invalid callback'}"
|
|
assert source.count(original) == 1
|
|
source = source.replace(original, "if($uri.Query){throw 'fixture query refused'}")
|
|
source = source.rstrip()
|
|
assert source.endswith("}")
|
|
source = source[:-1] + "\n[Console]::Out.WriteLine((@{fixture_complete=$true;opened=$opened;private_profile_removed=$removed}|ConvertTo-Json -Compress))\n}"
|
|
command = [native.POWERSHELL, "-NoProfile", "-NonInteractive", "-EncodedCommand",
|
|
base64.b64encode(source.encode("utf-16le")).decode()]
|
|
result = subprocess.run(command, input=payload, capture_output=True, timeout=45, check=False)
|
|
lines = result.stdout.decode("utf-8", errors="replace").splitlines()
|
|
assert lines[0] == "opened"
|
|
metadata = json.loads(lines[-1])
|
|
assert metadata == {"fixture_complete": True, "opened": True, "private_profile_removed": True}
|
|
assert "/" in visits
|
|
assert "/ready" in visits
|
|
print(json.dumps({"status": "passed", "native_identity_request": False,
|
|
"production_guard_refused_synthetic_url": True, "worker_url_received_on_stdin": True,
|
|
"windows_loopback_reached": True, "private_acl_verified_before_launch": True,
|
|
"windows_localhost_redirect_reached": True,
|
|
"owned_process_closed": True, "private_profile_removed": True}))
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join(timeout=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|