tegwick
f6eec13f7c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1abf1-7e95-7141-87c9-ee78332bf790
236 lines
12 KiB
Python
236 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Credential-free manual UI probe using the real operator Chromium launcher.
|
|
|
|
Default reviews only. --run opens a fresh window and private profile against a
|
|
disposable loopback page. --software-x11 selects X11/software rendering for
|
|
comparison. No issuer URL, account, credentials, callbacks or real profile.
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
LAUNCHER = Path(__file__).resolve().parents[2] / "railiance-platform/scripts/operator-browser/xdg-open"
|
|
POWERSHELL = "/mnt/c/Windows/System32/WindowsPowerShell/v1.0/powershell.exe"
|
|
WINDOWS_PROBE = r'''
|
|
$ErrorActionPreference='Stop'
|
|
$ProgressPreference='SilentlyContinue'
|
|
$profile=$null;$process=$null;$startTicks=$null;$closed=$false;$removed=$false;$userClosed=$false;$ok=$false
|
|
$stage='browser_detection';$failureStage=$null;$errorKind=$null
|
|
try {
|
|
$browser=Join-Path ${env:ProgramFiles(x86)} 'Microsoft\Edge\Application\msedge.exe'
|
|
if(-not (Test-Path -LiteralPath $browser)){throw 'browser unavailable'}
|
|
$stage='profile_creation'
|
|
$profile=Join-Path ([IO.Path]::GetTempPath()) ('railiance-browser-probe-'+[guid]::NewGuid().ToString('N'))
|
|
New-Item -ItemType Directory -Path $profile | Out-Null
|
|
$stage='profile_acl'
|
|
$stage='profile_acl_identity'
|
|
$sid=[Security.Principal.WindowsIdentity]::GetCurrent().User
|
|
$stage='profile_acl_apply'
|
|
$grant='*'+$sid.Value+':(OI)(CI)F'
|
|
& "$env:SystemRoot\System32\icacls.exe" $profile '/grant:r' $grant '/inheritance:r' | Out-Null
|
|
if($LASTEXITCODE -ne 0){throw 'private profile ACL command failed'}
|
|
$stage='profile_acl_verify'
|
|
$verified=Get-Acl -LiteralPath $profile
|
|
$rules=$verified.GetAccessRules($true,$true,[Security.Principal.SecurityIdentifier])
|
|
if(-not $verified.AreAccessRulesProtected -or $rules.Count -ne 1 -or $rules[0].IdentityReference.Value -ne $sid.Value){throw 'private profile ACL verification failed'}
|
|
$arguments=@(('--user-data-dir="'+$profile+'"'),'--no-first-run','--no-default-browser-check','--disable-sync','--new-window','__URL__')
|
|
$stage='process_start'
|
|
$process=Start-Process -FilePath $browser -ArgumentList $arguments -PassThru
|
|
$startTicks=$process.StartTime.ToUniversalTime().Ticks
|
|
$stage='process_wait'
|
|
$deadline=[datetime]::UtcNow.AddSeconds(__MAXIMUM__)
|
|
while(-not $process.HasExited -and [datetime]::UtcNow -lt $deadline){Start-Sleep -Milliseconds 250;$process.Refresh()}
|
|
$userClosed=$process.HasExited;$ok=$true
|
|
} catch {
|
|
$ok=$false;$failureStage=$stage;$errorKind=$_.Exception.GetType().Name
|
|
} finally {
|
|
if($process -and $startTicks){
|
|
$current=Get-Process -Id $process.Id -ErrorAction SilentlyContinue
|
|
if($current -and $current.StartTime.ToUniversalTime().Ticks -eq $startTicks -and $current.Path -eq $browser){
|
|
$null=$current.CloseMainWindow()
|
|
if(-not $current.WaitForExit(5000)){Stop-Process -Id $current.Id -Force -ErrorAction SilentlyContinue;$null=$current.WaitForExit(5000)}
|
|
}
|
|
$process.Refresh();$closed=$process.HasExited
|
|
}
|
|
if($profile -and ($closed -or -not $process)){
|
|
try {Remove-Item -LiteralPath $profile -Recurse -Force -ErrorAction Stop;$removed=$true} catch {$removed=$false}
|
|
}
|
|
@{status=$(if($ok){'native_probe_finished'}else{'native_probe_failed'});failure_stage=$failureStage;error_kind=$errorKind;browser='Windows Edge';user_closed=$userClosed;owned_process_closed=$closed;private_profile_removed=$removed;native_login=$false} | ConvertTo-Json -Compress
|
|
}
|
|
'''
|
|
HTML = b'''<!doctype html><meta charset="utf-8"><title>Browser response check - no login</title>
|
|
<style>body{font:20px system-ui;max-width:40rem;margin:2rem auto}button{display:block;margin:1rem;padding:1rem;background:#def;color:#123;width:32rem;text-align:left}output{display:block;padding:1rem;background:#eee}</style>
|
|
<h1>Browser response check</h1><p>This is a local test. No sign-in or credentials.</p>
|
|
<button id="one">1. First test button</button>
|
|
<button id="two">2. Second test button</button>
|
|
<button id="three">3. Third test button</button>
|
|
<output id="result">No test button activated yet.</output>
|
|
<p>Click each numbered button and check that the reported number matches.
|
|
Then press Tab to focus a button and Enter to activate it. Close this window with its titlebar X when done.</p>
|
|
<script>
|
|
for(const button of document.querySelectorAll('button'))button.addEventListener('click',event=>{
|
|
document.getElementById('result').textContent='Activated: '+button.textContent;
|
|
const r=button.getBoundingClientRect();
|
|
const observation={target:button.id,keyboard:event.detail===0,client_x:event.clientX,client_y:event.clientY,
|
|
device_scale:window.devicePixelRatio,button_top:r.top,button_bottom:r.bottom};
|
|
fetch('/event',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(observation)});
|
|
});
|
|
</script>
|
|
'''
|
|
|
|
|
|
def launcher_module():
|
|
loader = importlib.machinery.SourceFileLoader("operator_browser_probe_launcher", str(LAUNCHER))
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def probe_argv(module, browser, profile, url, software_x11):
|
|
command = module.launch_argv(browser, str(profile), url)
|
|
if software_x11:
|
|
command[-1:-1] = ["--ozone-platform=x11", "--disable-gpu"]
|
|
return command
|
|
|
|
|
|
def windows_argv(url, maximum):
|
|
parsed = urlsplit(url)
|
|
if not (parsed.scheme == "http" and parsed.hostname == "127.0.0.1" and parsed.port
|
|
and parsed.path == "/" and not parsed.query and not parsed.fragment
|
|
and not parsed.username and not parsed.password and 10 <= maximum <= 180):
|
|
raise ValueError("synthetic_loopback_only")
|
|
script = WINDOWS_PROBE.replace("__URL__", url).replace("__MAXIMUM__", str(maximum - 10))
|
|
encoded = base64.b64encode(script.encode("utf-16le")).decode()
|
|
return [POWERSHELL, "-NoProfile", "-NonInteractive", "-EncodedCommand", encoded]
|
|
|
|
|
|
def run(software_x11, maximum, windows=False):
|
|
if not windows:
|
|
module = launcher_module()
|
|
browser = module.browser_path()
|
|
if not browser:
|
|
raise RuntimeError("pinned_browser_unavailable")
|
|
events = []
|
|
started = time.monotonic()
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
if self.path not in ("/", "/favicon.ico"):
|
|
self.send_error(404)
|
|
return
|
|
if self.path != "/favicon.ico":
|
|
event = {"event": self.path, "elapsed_seconds": round(time.monotonic() - started, 2)}
|
|
events.append(event)
|
|
print(json.dumps(event), flush=True)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; connect-src 'self'; frame-ancestors 'none'")
|
|
self.end_headers()
|
|
self.wfile.write(HTML)
|
|
|
|
def do_POST(self):
|
|
if self.path != "/event" or not 0 < int(self.headers.get("Content-Length", 0)) <= 1024:
|
|
self.send_error(400)
|
|
return
|
|
try:
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
if set(body) != {"target", "keyboard", "client_x", "client_y", "device_scale", "button_top", "button_bottom"}:
|
|
raise ValueError
|
|
if body["target"] not in ("one", "two", "three") or type(body["keyboard"]) is not bool:
|
|
raise ValueError
|
|
if any(type(body[k]) not in (int, float) for k in body if k not in ("target", "keyboard")):
|
|
raise ValueError
|
|
except (ValueError, KeyError, TypeError):
|
|
self.send_error(400)
|
|
return
|
|
event = {"event": "button_activated", "elapsed_seconds": round(time.monotonic() - started, 2), **body}
|
|
events.append(event)
|
|
print(json.dumps(event), flush=True)
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="vergabe-browser-probe-") as temporary:
|
|
root = Path(temporary)
|
|
home, profile = root / "empty-home", root / "empty-profile"
|
|
home.mkdir(mode=0o700)
|
|
profile.mkdir(mode=0o700)
|
|
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) + "/"
|
|
command = windows_argv(url, maximum) if windows else probe_argv(module, browser, profile, url, software_x11)
|
|
env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")}
|
|
env["HOME"] = str(home)
|
|
process = subprocess.Popen(command, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE if windows else subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, start_new_session=True)
|
|
print(json.dumps({"status": "ui_probe_started", "software_x11": software_x11,
|
|
"native_windows": windows, "native_login": False, "maximum_seconds": maximum}), flush=True)
|
|
try:
|
|
while process.poll() is None and time.monotonic() - started < maximum:
|
|
time.sleep(0.25)
|
|
finally:
|
|
user_closed = process.poll() is not None
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=8)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join(timeout=2)
|
|
if windows:
|
|
raw = process.stdout.read(16385)
|
|
process.stdout.close()
|
|
try:
|
|
metadata = json.loads(raw) if len(raw) <= 16384 else {}
|
|
print(json.dumps({k: metadata.get(k) for k in ("status", "failure_stage", "error_kind", "browser", "user_closed", "owned_process_closed", "private_profile_removed", "native_login")}), flush=True)
|
|
except (ValueError, UnicodeError):
|
|
print(json.dumps({"status": "native_probe_metadata_unavailable", "details_suppressed": True}), flush=True)
|
|
print(json.dumps({"status": "ui_probe_complete", "software_x11": software_x11,
|
|
"window_process_exited_before_deadline": user_closed, "events": events,
|
|
"human_wait_included_in_times": True, "native_login": False}), flush=True)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--run", action="store_true")
|
|
parser.add_argument("--software-x11", action="store_true")
|
|
parser.add_argument("--windows", action="store_true", help="use native Windows Edge with its own disposable profile")
|
|
parser.add_argument("--maximum-seconds", type=int, default=120)
|
|
args = parser.parse_args()
|
|
if not 10 <= args.maximum_seconds <= 180:
|
|
parser.error("maximum must be between 10 and 180 seconds")
|
|
if args.windows and args.software_x11:
|
|
parser.error("Windows and Linux rendering modes are mutually exclusive")
|
|
if not args.run:
|
|
print(json.dumps({"status": "prepared", "native_login": False, "fresh_profile": True,
|
|
"software_x11": args.software_x11, "native_windows": args.windows, "next": "--run"}))
|
|
return 0
|
|
run(args.software_x11, args.maximum_seconds, args.windows)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
os.umask(0o077)
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception: # noqa: BLE001 — do not return browser or environment diagnostics
|
|
print(json.dumps({"status": "ui_probe_failed", "details_suppressed": True}))
|
|
raise SystemExit(1)
|