admin_rb
201 lines
15 KiB
Python
201 lines
15 KiB
Python
"""Chromium about:blank fixtures bridged to real ASGI endpoints; synthetic worker handoffs.
|
|
|
|
This is not live systemd/Core/SSH qualification. Bootstrap may be a declared
|
|
substitute. HTMX is deliberately not loaded when unavailable: status replacements
|
|
are driven explicitly and labeled as synthetic lifecycle coverage, not HTMX proof.
|
|
Never install fixture assets into production static/vendor.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from urllib.parse import urlsplit
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'src'))
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from playwright.sync_api import sync_playwright
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.credentials import wire
|
|
from test_credential_ux import authz as auth_fixture, make_job, SYNTHETIC
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--out', type=Path, required=True)
|
|
parser.add_argument('--bootstrap', type=Path, default=Path('/opt/imagemagick/share/doc/ImageMagick-7/www/assets/bootstrap.min.css'))
|
|
args = parser.parse_args(); args.out.mkdir(parents=True, exist_ok=True)
|
|
cases = []
|
|
with tempfile.TemporaryDirectory(prefix='aim-credential-qa-') as tmp, pytest.MonkeyPatch.context() as monkey:
|
|
auth, users = auth_fixture.__wrapped__(Path(tmp), monkey)
|
|
fixture = (auth, users)
|
|
packets = []
|
|
class Sock:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
monkey.setattr(wire, 'connect', lambda *a, **k: Sock())
|
|
monkey.setattr(wire, 'send', lambda sock, packet, limit: packets.append(json.loads(json.dumps(packet))))
|
|
def claimed(*args):
|
|
ident = packets[-1]['job']
|
|
with auth.store.transaction() as db:
|
|
db.execute("UPDATE jobs SET credential_phase='claimed' WHERE id=?", (ident,))
|
|
return {'accepted': True}
|
|
monkey.setattr(wire, 'receive', claimed)
|
|
origin = auth.settings.public_url
|
|
with TestClient(create_app(auth.settings), base_url=origin, follow_redirects=False) as server, sync_playwright() as pw:
|
|
browser = pw.chromium.launch(executable_path='/usr/bin/chromium', args=['--no-sandbox', '--disable-dev-shm-usage'])
|
|
try:
|
|
for width, height in ((320,750),(390,844),(760,900),(1440,1000),(740,390)):
|
|
for theme in ('light','dark'):
|
|
# Independent viewport fixtures must not share the live five/minute bucket.
|
|
# Production throttle remains enabled and is covered in unit tests.
|
|
with auth.store.transaction() as db: db.execute('DELETE FROM rate_limits')
|
|
token, session = auth.new_session(users['operator'])
|
|
context = browser.new_context(viewport={'width':width,'height':height}, locale='en-GB', timezone_id='Europe/Berlin', reduced_motion='reduce')
|
|
context.add_cookies([{'name':auth.settings.cookie_name, 'value':token, 'url':origin,'secure':True,'httpOnly':True,'sameSite':'Lax'}])
|
|
errors = []
|
|
mode = {'drop_post':False}
|
|
def bridge(url, options):
|
|
path = urlsplit(url).path
|
|
method = options.get('method', 'GET')
|
|
if mode['drop_post'] and method == 'POST' and path.endswith('/credentials'):
|
|
mode['post_count'] = mode.get('post_count',0)+1
|
|
return {'network_error': True}
|
|
headers = {**options.get('headers', {}), 'Origin': origin}
|
|
response = server.request(method, path, headers=headers, content=options.get('body'))
|
|
return {'status': response.status_code, 'body': response.text}
|
|
server.cookies.set(auth.settings.cookie_name, token)
|
|
page = context.new_page(); page.on('pageerror', lambda e: errors.append(str(e)))
|
|
page.expose_function('_fixtureHTTP', bridge)
|
|
def mount(path):
|
|
page.goto('about:blank')
|
|
content = server.get(path).text
|
|
content = re.sub(r'<script[^>]+src="[^"]+"[^>]*></script>', '', content)
|
|
content = re.sub(r'<link[^>]+rel="stylesheet"[^>]*>', '', content)
|
|
css = args.bootstrap.read_text() + '\n' + '\n'.join((ROOT/'src/aim_webgui/static/css'/n).read_text() for n in ('tokens.css','bootstrap-overrides.css','aim.css','experience.css','credentials.css','evidence.css'))
|
|
fixture_js = """
|
|
window.fetch = async function(url, options) {
|
|
const r=await window._fixtureHTTP(url, options||{});
|
|
if(r.network_error) throw new TypeError('Synthetic lost response');
|
|
return new Response(r.body,{status:r.status,headers:{'Content-Type':'application/json'}});
|
|
};
|
|
window.EventSource=class { constructor(){} addEventListener(){} close(){} };
|
|
"""
|
|
js='\n'.join((ROOT/'src/aim_webgui/static/js'/n).read_text() for n in ('theme.js','aim.js','experience.js','credentials.js','evidence.js'))
|
|
content=content.replace('</head>','<style>'+css+'</style></head>').replace('</body>','<script>'+fixture_js+js+'</script></body>')
|
|
page.set_content(content, wait_until='load')
|
|
try:
|
|
ident = make_job(fixture)
|
|
mount('/jobs/'+ident)
|
|
page.locator('[data-theme-option='+theme+']:visible').click()
|
|
opener = page.locator('[data-credential-open]')
|
|
opener.click()
|
|
panel = page.locator('dialog [data-credential-panel]'); panel.wait_for()
|
|
assert page.locator('dialog').evaluate('(e)=>e.open')
|
|
assert page.locator('#credential-dialog-title').evaluate('(e)=>e===document.activeElement')
|
|
assert page.locator('[data-key-source-choice]').is_visible()
|
|
assert not page.locator('[name="ssh_key_passphrase"]').is_visible()
|
|
assert not page.locator('dialog').evaluate('(e)=>e.scrollWidth>e.clientWidth+1')
|
|
assert page.evaluate('document.documentElement.scrollWidth<=innerWidth')
|
|
box=page.locator('dialog').bounding_box(); assert box['x']>=0 and box['y']>=0 and box['x']+box['width']<=width+1 and box['y']+box['height']<=height+1
|
|
# Native modal keeps background inert and cycles focus within dialog.
|
|
for _ in range(12):
|
|
page.keyboard.press('Tab')
|
|
assert page.evaluate("document.querySelector('dialog').contains(document.activeElement)")
|
|
page.locator('[name="vault_password"]').fill(SYNTHETIC)
|
|
page.locator('button[aria-controls="credential-vault"]').click()
|
|
assert page.locator('[name="vault_password"]').get_attribute('type')=='text'
|
|
# Poll replacement cannot replace the modal or its typed/revealed input.
|
|
page.evaluate("async (job)=>{const r=await fetch('/_partials/jobs/'+job);const html=await r.text();document.getElementById('job-status').outerHTML=html;document.dispatchEvent(new CustomEvent('htmx:afterSwap',{detail:{target:document.getElementById('job-status')}}));}", ident)
|
|
assert page.locator('[name="vault_password"]').input_value()==SYNTHETIC
|
|
page.locator('label[for="credential-key-separate"]').click()
|
|
page.locator('[name="ssh_key_passphrase"]').fill('Synthetic separate key')
|
|
page.locator('label[for="credential-key-vault"]').click()
|
|
assert page.locator('[name="ssh_key_passphrase"]').input_value()==''
|
|
# Revealed input must be cleared on Escape, not just hidden visually.
|
|
page.keyboard.press('Escape')
|
|
assert not page.locator('dialog').evaluate('(e)=>e.open')
|
|
assert page.locator('[data-credential-secret]').count()==0
|
|
assert page.locator('[data-credential-open]').evaluate('(e)=>e===document.activeElement')
|
|
page.locator('[data-credential-open]').click(); panel.wait_for()
|
|
assert page.locator('[name="vault_password"]').input_value()==''
|
|
if theme=='dark' and width in (390,1440):
|
|
page.screenshot(path=str(args.out/f'credential-dialog-{width}.png'), full_page=False)
|
|
start=len(packets)
|
|
page.locator('[name="vault_password"]').fill(SYNTHETIC)
|
|
page.locator('[data-credential-submit]').click()
|
|
page.locator('[data-credential-feedback][data-tone="accepted"]').wait_for()
|
|
assert len(packets)==start+1 and packets[-1]['credentials']['vault_password']==SYNTHETIC
|
|
assert 'ssh_key_passphrase' not in packets[-1]['credentials']
|
|
assert page.locator('[name="vault_password"]').input_value()==''
|
|
assert 'accepted' in page.locator('[data-credential-feedback]').inner_text().lower()
|
|
assert page.url=='about:blank'
|
|
page.locator('[data-credential-dismiss]').click()
|
|
assert not page.locator('dialog').evaluate('(e)=>e.open')
|
|
# Attention action belongs to this viewer; choose only their credentials.
|
|
ident2=make_job(fixture, requirements=['vault_password'])
|
|
mount('/jobs')
|
|
page.locator('#attention [data-job-id="'+ident2+'"]').click(); panel.wait_for()
|
|
assert page.locator('[data-key-source-choice]').count()==0
|
|
# Lost acknowledgement: only one POST, then GET reconciliation; no replay.
|
|
mode['drop_post']=True;mode['post_count']=0
|
|
page.locator('[name="vault_password"]').fill(SYNTHETIC)
|
|
page.locator('[data-credential-submit]').click()
|
|
page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('not confirmed')")
|
|
page.wait_for_timeout(3500)
|
|
assert mode['post_count']==1 and page.locator('[name="vault_password"]').input_value()==''
|
|
page.locator('[data-credential-dismiss]').click()
|
|
page.locator('#attention [data-job-id="'+ident2+'"]').click();panel.wait_for()
|
|
page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('not been confirmed')")
|
|
assert page.locator('[data-credential-submit]').is_hidden()
|
|
page.keyboard.press('Escape');mode['drop_post']=False
|
|
# Server cancels while the form is open: input is cleared without POST.
|
|
ident3=make_job(fixture)
|
|
mount('/jobs/'+ident3);page.locator('[data-credential-open]').click();panel.wait_for()
|
|
page.locator('[name="vault_password"]').fill(SYNTHETIC)
|
|
with auth.store.transaction() as db: db.execute('UPDATE jobs SET cancel_requested=1 WHERE id=?',(ident3,))
|
|
page.wait_for_function("document.querySelector('[data-credential-feedback]').textContent.includes('cancellation')")
|
|
assert page.locator('[name="vault_password"]').input_value()==''
|
|
page.keyboard.press('Escape')
|
|
# Short visual viewport is internally scrollable; footer stays reachable.
|
|
ident4=make_job(fixture)
|
|
mount('/jobs/'+ident4);page.locator('[data-credential-open]').click();panel.wait_for()
|
|
page.set_viewport_size({'width':width,'height':min(360,height)})
|
|
page.wait_for_timeout(100)
|
|
assert page.locator('.credential-dialog-body').evaluate('(e)=>e.scrollHeight>e.clientHeight')
|
|
page.locator('[data-credential-submit]').scroll_into_view_if_needed()
|
|
box=page.locator('dialog').bounding_box();assert box['y']>=-1 and box['y']+box['height']<=min(360,height)+1
|
|
# Page hide clears revealed secrets too. BFCache style re-init never restores them.
|
|
page.locator('[name="vault_password"]').fill(SYNTHETIC)
|
|
page.locator('button[aria-controls="credential-vault"]').click()
|
|
page.evaluate("window.dispatchEvent(new PageTransitionEvent('pagehide'))")
|
|
assert page.locator('[data-credential-secret]').count()==0
|
|
assert not errors, errors
|
|
cases.append({'width':width,'height':height,'theme':theme,'passed':True})
|
|
print('Credential QA passed',width,height,theme,flush=True)
|
|
finally: context.close()
|
|
# No-JS native markup and a plain HTTP form POST are verified separately.
|
|
with auth.store.transaction() as db: db.execute('DELETE FROM rate_limits')
|
|
token, session = auth.new_session(users['operator'])
|
|
server.cookies.set(auth.settings.cookie_name, token)
|
|
ident=make_job(fixture,requirements=['vault_password'])
|
|
r=server.get('/jobs/'+ident+'/credentials')
|
|
assert r.status_code==200 and 'method="post"' in r.text
|
|
r=server.post('/jobs/'+ident+'/credentials', data={'_csrf':session['csrf'],'vault_password':SYNTHETIC},headers={'Origin':origin})
|
|
assert r.status_code==303 and r.headers['location']=='/jobs/'+ident
|
|
cases.append({'plain_http_form_fallback':True,'passed':True})
|
|
finally: browser.close()
|
|
report={'fixture_only':True,'cases':cases,'asgi_auth_csrf_endpoints':True,'browser_network':'about:blank fetch bridge to real TestClient; no real browser CSP/TLS/cookie transport qualification',
|
|
'bootstrap_source':str(args.bootstrap),'bootstrap_substitution':'5.3.6, production pin 5.3.8',
|
|
'htmx_loaded':False,'htmx_swap':'synthetic replacement + lifecycle event',
|
|
'real_worker_core_execution':False,'browser':'Chromium via Playwright'}
|
|
(args.out/'results.json').write_text(json.dumps(report,indent=2))
|
|
print(len(cases),'browser scenario groups passed')
|
|
|
|
if __name__=='__main__': main()
|