key-cape/scripts/provider_browser_acceptance.py

#!/usr/bin/env python3
"""Run installed-provider browser and KeyCape adapter acceptance in a disposable Job.

No production environment, service account, databases or real users are mounted.
Requires kubectl, Chromium, Node and Go; prints checks, never OTP material.
"""
from pathlib import Path
import json
import os
import shutil
import socket
import subprocess
import tempfile
import time
import urllib.request
import uuid

ROOT=Path(__file__).resolve().parents[1]
IMAGE='ghcr.io/gpappsoft/privacyidea-docker@sha256:af7841adad262f129e0c1d4f553af13f21cb2f4dc713533f316cfe43ed0b4473'

def main():
    chrome=os.environ.get('JOURNEY_CHROME') or shutil.which('chromium') or shutil.which('google-chrome')
    if not chrome:
        candidates=sorted((Path.home()/'.cache/ms-playwright').glob('chromium-*/chrome-linux64/chrome'))
        chrome=str(candidates[-1]) if candidates else None
    if not chrome or not shutil.which('node') or not shutil.which('go'):
        raise RuntimeError('Chromium, Node and Go are required; acceptance was not run')
    guard=(ROOT/'scripts/keycape_onboarding_guard.py').read_text()
    source='import sys,types\ng=types.ModuleType("keycape_onboarding_guard");sys.modules["keycape_onboarding_guard"]=g\nexec('+repr(guard)+',g.__dict__)\n'+(ROOT/'scripts/provider-browser-fixture.py').read_text()
    name='provider-p06-browser-'+uuid.uuid4().hex[:8]
    job={'apiVersion':'batch/v1','kind':'Job','metadata':{'name':name,'namespace':'mfa'},'spec':{'activeDeadlineSeconds':600,'ttlSecondsAfterFinished':3600,'template':{'metadata':{'labels':{'app.kubernetes.io/name':'provider-p06-browser'}},'spec':{'automountServiceAccountToken':False,'restartPolicy':'Never','securityContext':{'runAsUser':65534,'runAsGroup':65534,'runAsNonRoot':True,'fsGroup':65534},'containers':[{'name':'fixture','image':IMAGE,'command':['python3','-c',source],'env':[{'name':'PYTHONDONTWRITEBYTECODE','value':'1'}],'securityContext':{'allowPrivilegeEscalation':False,'readOnlyRootFilesystem':True,'capabilities':{'drop':['ALL']}},'resources':{'requests':{'cpu':'1m','memory':'128Mi'},'limits':{'cpu':'500m','memory':'384Mi'}},'volumeMounts':[{'name':'scratch','mountPath':'/tmp'}]}],'volumes':[{'name':'scratch','emptyDir':{}}]}}}}
    created=subprocess.run(['kubectl','create','-f','-'],input=json.dumps(job).encode(),capture_output=True)
    if created.returncode:raise RuntimeError('Fixture creation failed')
    forward=None
    forward_log=tempfile.TemporaryFile(mode="w+t")
    try:
        for _ in range(120):
            result=subprocess.run(['kubectl','-n','mfa','get','pods','-l','job-name='+name,'-o','json'],capture_output=True)
            pods=json.loads(result.stdout).get('items',[]) if result.returncode==0 else []
            if pods and pods[0]['status']['phase']=='Running':break
            time.sleep(.5)
        else:raise RuntimeError('Fixture did not start')
        # A Running pod is not an HTTP-ready provider. Connecting too early can
        # terminate kubectl port-forward on the first refused upstream socket.
        for _ in range(120):
            logs=subprocess.run(['kubectl','-n','mfa','logs','job/'+name],capture_output=True,text=True)
            events=[]
            for line in logs.stdout.splitlines():
                try: events.append(json.loads(line))
                except ValueError: pass
            if any(event.get('fixture_ready') is True for event in events if isinstance(event,dict)):break
            if any('fixture_failure' in event for event in events if isinstance(event,dict)):raise RuntimeError('Fixture initialization failed')
            time.sleep(.5)
        else:raise RuntimeError('Fixture initialization timed out')
        with socket.socket() as listener:listener.bind(('127.0.0.1',0));port=listener.getsockname()[1]
        forward=subprocess.Popen(['kubectl','-n','mfa','port-forward','job/'+name,str(port)+':8088','--address=127.0.0.1'],stdout=subprocess.DEVNULL,stderr=forward_log)
        base='http://127.0.0.1:'+str(port)
        for _ in range(120):
            if forward.poll() is not None:raise RuntimeError('Fixture forwarding stopped')
            try:
                with urllib.request.urlopen(base+'/',timeout=2) as response:
                    if response.status==200:break
            except Exception:time.sleep(.25)
        else:raise RuntimeError('Fixture HTTP unavailable')
        environment=dict(os.environ,P06_NATIVE_PROVIDER_URL=base)
        subprocess.run(['go','test','./internal/server/oidc','-run','^TestNativeOptionalEnrollmentAndOldSession$','-count=1','-v'],cwd=ROOT/'src',env=environment,check=True,timeout=120)
        with tempfile.TemporaryDirectory(prefix='p06-provider-browser-') as profile:
            browser=subprocess.Popen([chrome,'--headless','--no-sandbox','--disable-gpu','--remote-debugging-address=127.0.0.1','--remote-debugging-port=0','--user-data-dir='+profile,'about:blank'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
            try:
                marker=Path(profile)/'DevToolsActivePort'
                for _ in range(100):
                    if marker.exists():break
                    time.sleep(.1)
                debug='http://127.0.0.1:'+marker.read_text().splitlines()[0]
                subprocess.run(['node',str(ROOT/'scripts/provider_browser.mjs'),debug,base],check=True,timeout=240)
            finally:browser.terminate();browser.wait(timeout=5)
        print(json.dumps({'success':True,'job':name,'provider_image':IMAGE,'production_accounts_changed':False}))
    except Exception:
        if forward and forward.poll() is not None:
            forward_log.seek(0)
            message=forward_log.read(1000)
            print(json.dumps({"forward_exit":forward.returncode,"reason": "pod_not_running" if "not running" in message else "permission" if "permission" in message.lower() else "port_in_use" if "address already in use" in message else "forward_failed"}),flush=True)
        logs=subprocess.run(['kubectl','-n','mfa','logs','job/'+name],capture_output=True,text=True)
        for line in logs.stdout.splitlines():
            try:
                event=json.loads(line)
                if 'fixture_failure' in event:print(json.dumps({k:event[k] for k in ['fixture_failure','phase']}),flush=True)
            except Exception:pass
        raise
    finally:
        if forward:forward.terminate();forward.wait(timeout=5)
        forward_log.close()
        subprocess.run(['kubectl','-n','mfa','delete','job',name,'--wait=false'],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)

if __name__=='__main__':main()