admin_rb
290 lines
15 KiB
Python
290 lines
15 KiB
Python
"""Credential modal/read status/attention boundaries; synthetic secrets, no Core run."""
|
|
from dataclasses import replace
|
|
import json
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.auth.service import Auth
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui.workflows import Workflows
|
|
from aim_webgui.credentials import presentation, service, wire
|
|
from aim_webgui.errors import WebError
|
|
|
|
SYNTHETIC = 'fixture-only +%! {{ 7 * 7 }} " snowman \u2603'
|
|
|
|
|
|
@pytest.fixture
|
|
def authz(tmp_path, monkeypatch):
|
|
settings = Settings(state_dir=tmp_path/'state', public_url='https://aim.example.test',
|
|
execution_enabled=True, execution_require_approval=False,
|
|
execution_playbooks=('checkmk_install_agent',),
|
|
execution_transport_verified=True, credentials_enabled=True).validate()
|
|
auth = Auth(settings); auth.bootstrap()
|
|
for name, role in [('operator', 'viewer'), ('another', 'viewer'), ('admin2', 'admin')]:
|
|
auth.create_user(name, 'Synthetic-fixture-password-2026', role)
|
|
with auth.store.transaction() as db:
|
|
db.execute('UPDATE users SET must_change_password=0')
|
|
users = {row['username']: row['id'] for row in db.execute('SELECT id,username FROM users')}
|
|
for user in ('operator', 'another'):
|
|
db.execute('INSERT INTO grants(user_id,customer,playbook) VALUES(?,?,?)',
|
|
(users[user], 'example', 'checkmk_install_agent'))
|
|
# New UX must not prepare/execute/decrypt/probe remotely on page load.
|
|
from aim_webgui.adapters.core_v1 import CoreAdapter
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('Credential UX must not call Core')
|
|
for name in ('preflight', 'readiness', 'reprepare'):
|
|
monkeypatch.setattr(CoreAdapter, name, forbidden)
|
|
return auth, users
|
|
|
|
|
|
def make_job(authz, owner='operator', *, requirements=None, status='running', phase='waiting', deadline=None, canceled=False, when=None):
|
|
auth, users = authz
|
|
now = int(time.time())
|
|
ident = uuid.uuid4().hex
|
|
plan = {'customer': 'example', 'playbook': 'checkmk_install_agent', 'targets': ['lab.example'],
|
|
'overrides': {}, 'core_request': {'key_mode': 'customer'},
|
|
'credential_requirements': requirements if requirements is not None else ['vault_password', 'ssh_key_passphrase_or_customer_vault_value']}
|
|
with auth.store.transaction() as db:
|
|
db.execute('''INSERT INTO jobs(id,owner_id,plan,mode,status,created_at,scheduled_at,
|
|
credential_phase,credential_deadline,cancel_requested)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?)''',
|
|
(ident, users[owner], json.dumps(plan), 'apply', status, when or now, now, phase,
|
|
now+240 if deadline is None else deadline, canceled))
|
|
return ident
|
|
|
|
|
|
def client(authz, user='operator'):
|
|
auth, users = authz
|
|
token, session = auth.new_session(users[user])
|
|
browser = TestClient(create_app(auth.settings), base_url=auth.settings.public_url, follow_redirects=False)
|
|
browser.cookies.set(auth.settings.cookie_name, token)
|
|
browser.headers.update({'Origin': auth.settings.public_url, 'X-CSRF-Token': session['csrf']})
|
|
return browser
|
|
|
|
|
|
def test_modal_fragment_is_eligible_owner_only_and_full_page_fallback(authz):
|
|
ident = make_job(authz)
|
|
with client(authz) as b:
|
|
r = b.get('/jobs/'+ident)
|
|
assert r.status_code == 200
|
|
assert 'data-credential-open' in r.text and '<dialog' in r.text
|
|
assert 'name="vault_password"' not in r.text # Lazy-loaded, outside pollable fragment.
|
|
r = b.get('/_partials/jobs/'+ident+'/credentials')
|
|
assert r.status_code == 200
|
|
assert 'Use customer Vault' in r.text and 'Enter separately' in r.text
|
|
assert 'type="radio"' in r.text and 'btn-check' in r.text
|
|
assert 'data-credential-secret' in r.text and 'hx-history="false"' in r.text
|
|
assert 'value="'+SYNTHETIC+'"' not in r.text
|
|
assert 'no-store' in r.headers['cache-control']
|
|
r = b.get('/jobs/'+ident+'/credentials')
|
|
assert r.status_code == 200 and 'data-credential-form' in r.text
|
|
assert 'action="/jobs/'+ident+'/credentials"' in r.text
|
|
for user in ('another', 'admin2'):
|
|
with client(authz, user) as b:
|
|
assert b.get('/_partials/jobs/'+ident+'/credentials').status_code in (403, 404)
|
|
assert b.get('/api/v2/runs/'+ident+'/credential-status').status_code in (403, 404)
|
|
|
|
|
|
def test_only_core_reported_fields_shown(authz):
|
|
vault = make_job(authz, requirements=['vault_password'])
|
|
key = make_job(authz, requirements=['ssh_key_passphrase_or_customer_vault_value'])
|
|
conn = make_job(authz, requirements=['connection_password'])
|
|
unsupported = make_job(authz, requirements=['become_password'])
|
|
with client(authz) as b:
|
|
v = b.get('/_partials/jobs/'+vault+'/credentials').text
|
|
assert 'name="vault_password"' in v and 'name="ssh_key_passphrase"' not in v
|
|
k = b.get('/_partials/jobs/'+key+'/credentials').text
|
|
assert 'Use customer Vault' not in k
|
|
assert 'name="ssh_key_passphrase" maxlength="2048" required' in k
|
|
c = b.get('/_partials/jobs/'+conn+'/credentials').text
|
|
assert 'name="connection_password"' in c and 'name="vault_password"' not in c
|
|
assert b.get('/_partials/jobs/'+unsupported+'/credentials').status_code == 409
|
|
|
|
|
|
@pytest.mark.parametrize('status,phase,canceled,expired', [
|
|
('queued', '', False, False), ('running', 'claimed', False, False),
|
|
('running', 'waiting', True, False), ('running', 'waiting', False, True),
|
|
('failed', 'released', False, False)])
|
|
def test_reservation_state_changes_never_keep_form_eligible(authz, status, phase, canceled, expired):
|
|
ident = make_job(authz, status=status, phase=phase, canceled=canceled,
|
|
deadline=int(time.time())-1 if expired else None)
|
|
with client(authz) as b:
|
|
r = b.get('/api/v2/runs/'+ident+'/credential-status')
|
|
assert r.status_code == 200 and not r.json()['can_submit']
|
|
assert b.get('/_partials/jobs/'+ident+'/credentials').status_code == 409
|
|
r = b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC})
|
|
assert r.status_code == 409 and SYNTHETIC not in r.text
|
|
|
|
|
|
def test_status_get_is_read_only_and_does_not_renew(authz):
|
|
ident = make_job(authz)
|
|
auth, users = authz
|
|
flow = Workflows(auth.settings)
|
|
before = flow.job(users['operator'], ident)
|
|
with client(authz) as b:
|
|
for _ in range(3):
|
|
r = b.get('/api/v2/runs/'+ident+'/credential-status')
|
|
assert r.json()['can_submit']
|
|
assert set(r.json()) == {'job_id','status','phase','can_submit','cancel_requested','server_now','deadline'}
|
|
b.get('/_partials/jobs/'+ident+'/credentials')
|
|
after = flow.job(users['operator'], ident)
|
|
assert before['credential_deadline'] == after['credential_deadline']
|
|
assert before['events'] == after['events'] and before['status'] == after['status']
|
|
|
|
|
|
def test_revoked_grant_and_session_denied(authz):
|
|
ident = make_job(authz)
|
|
auth, users = authz
|
|
with client(authz) as b:
|
|
with auth.store.transaction() as db:
|
|
db.execute('DELETE FROM grants WHERE user_id=?', (users['operator'],))
|
|
assert b.get('/api/v2/runs/'+ident+'/credential-status').status_code == 403
|
|
assert b.get('/_partials/jobs/'+ident+'/credentials').status_code == 403
|
|
assert b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC}).status_code == 403
|
|
with auth.store.transaction() as db: db.execute('DELETE FROM sessions')
|
|
assert b.get('/api/v2/runs/'+ident+'/credential-status').status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize('route', ['/api/v2/runs/', '/api/v2/jobs/'])
|
|
def test_submit_keeps_one_run_wire_and_literal_secret_semantics(authz, monkeypatch, route):
|
|
ident = make_job(authz)
|
|
auth, users = authz
|
|
packets = []
|
|
class Sock:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
monkeypatch.setattr(wire, 'connect', lambda *a, **k: Sock())
|
|
monkeypatch.setattr(wire, 'send', lambda sock, value, limit: packets.append(json.loads(json.dumps(value))))
|
|
monkeypatch.setattr(wire, 'receive', lambda *a: {'accepted': True})
|
|
with client(authz) as b:
|
|
r = b.post(route+ident+'/credentials', json={'vault_password': SYNTHETIC})
|
|
assert r.status_code == 202 and r.json()['accepted']
|
|
assert 'not remote authentication verification' in r.json()['notice']
|
|
assert SYNTHETIC not in r.text
|
|
assert packets[0]['credentials']['vault_password'] == SYNTHETIC
|
|
assert packets[0]['job'] == ident and packets[0]['user'] == users['operator']
|
|
assert SYNTHETIC.encode() not in auth.settings.database.read_bytes()
|
|
|
|
|
|
def test_submission_unknown_does_not_claim_it_was_not_received(authz, monkeypatch):
|
|
ident = make_job(authz)
|
|
monkeypatch.setattr(wire, 'connect', lambda *a, **k: (_ for _ in ()).throw(OSError('synthetic internal detail')))
|
|
with client(authz) as b:
|
|
r = b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC})
|
|
assert r.status_code == 409 and 'could not be confirmed' in r.json()['error']['message']
|
|
assert 'synthetic internal' not in r.text and SYNTHETIC not in r.text
|
|
|
|
|
|
def test_csrf_origin_body_limit_duplicate_and_forbidden_fields(authz):
|
|
ident = make_job(authz)
|
|
with client(authz) as b:
|
|
path = '/api/v2/runs/'+ident+'/credentials'
|
|
assert b.post(path, json={'vault_password': SYNTHETIC}, headers={'Origin':'https://other.invalid'}).status_code == 403
|
|
csrf = b.headers.pop('X-CSRF-Token')
|
|
assert b.post(path, json={'vault_password': SYNTHETIC}).status_code == 403
|
|
b.headers['X-CSRF-Token'] = csrf
|
|
assert b.post(path, json={'vault_password': 'x'*9000}).status_code == 413
|
|
for value in ({'vault_password': '\n'}, {'vault_password': 'x'*2049}, {'username': 'root'},
|
|
{'become_password': 'x'}, {'vault_password': SYNTHETIC, 'scope': 'all'}):
|
|
r = b.post(path, json=value)
|
|
assert r.status_code == 400 and SYNTHETIC not in r.text
|
|
r = b.post('/jobs/'+ident+'/credentials', content='_csrf='+csrf+'&vault_password=x&vault_password=y',
|
|
headers={'Content-Type': 'application/x-www-form-urlencoded'})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_required_key_without_vault_and_no_new_auth_override():
|
|
with pytest.raises(WebError): service.fields({}, ['ssh_key_passphrase_or_customer_vault_value'])
|
|
assert service.fields({'ssh_key_passphrase': SYNTHETIC}, ['ssh_key_passphrase_or_customer_vault_value']) == {'ssh_key_passphrase': SYNTHETIC}
|
|
with pytest.raises(WebError): service.fields({}, ['ssh_key_passphrase'])
|
|
with pytest.raises(WebError): service.fields({'ssh_key_passphrase': SYNTHETIC}, ['vault_password'])
|
|
|
|
|
|
def test_attention_scope_not_latest_100_and_no_auto_approval(authz):
|
|
auth, users = authz
|
|
own = make_job(authz, when=1)
|
|
others = make_job(authz, owner='another')
|
|
pending = make_job(authz, owner='another', status='pending', phase='')
|
|
admin_own = make_job(authz, owner='admin2', status='pending', phase='')
|
|
for _ in range(101): make_job(authz, status='failed', phase='released')
|
|
data = presentation.attention(Workflows(auth.settings), users['operator'])
|
|
assert data['total'] == 1 and data['items'][0]['id'] == own
|
|
admin_data = presentation.attention(Workflows(auth.settings), users['admin2'])
|
|
assert admin_data['total'] == 1 and admin_data['items'][0]['id'] == pending
|
|
assert others not in repr(admin_data) and admin_own not in repr(admin_data)
|
|
with client(authz) as b:
|
|
text = b.get('/_partials/attention').text
|
|
assert 'data-credential-open' in text and own in text and others not in text
|
|
r = b.get('/jobs')
|
|
assert 'Credentials needed' in r.text and own in r.text
|
|
assert Workflows(auth.settings).job(users['admin2'],pending)['status'] == 'pending'
|
|
|
|
|
|
def test_modal_is_outside_status_poll_and_no_custom_toggle():
|
|
root = Path(__file__).resolve().parents[1]/'src/aim_webgui'
|
|
partial = (root/'templates/partials/job.html').read_text()
|
|
assert '<dialog' not in partial and '<input' not in partial.replace('<input type="hidden" name="_csrf" value="{{ csrf }}">','')
|
|
js = (root/'static/js/credentials.js').read_text()
|
|
assert 'localStorage' not in js and 'sessionStorage' not in js
|
|
assert 'dialog.showModal()' in js and 'performance.now()' in js
|
|
assert 'data-credential-secret' in js and 'credentials: \'same-origin\'' in js
|
|
assert "'Use custom credentials'" not in js
|
|
|
|
|
|
def test_real_worker_socket_claim_is_single_use_and_never_persisted(authz, monkeypatch):
|
|
"""Exercise the actual existing worker wait/claim; never call Core execution."""
|
|
import threading
|
|
import aim_webgui.worker as worker_module
|
|
ident = make_job(authz)
|
|
auth, users = authz
|
|
worker = worker_module.Worker(auth.settings, Path('/synthetic-unused-config.toml'))
|
|
monkeypatch.setattr(worker_module, 'revalidate', lambda *args: None)
|
|
with auth.store.read() as db:
|
|
row = dict(db.execute('SELECT * FROM jobs WHERE id=?',(ident,)).fetchone())
|
|
result, errors = [], []
|
|
def wait():
|
|
try: result.append(worker.wait_credentials(row))
|
|
except Exception as error: errors.append(error)
|
|
thread = threading.Thread(target=wait, daemon=True);thread.start()
|
|
until = time.monotonic()+3
|
|
while not (auth.settings.state_dir/'.credential.sock').exists() and time.monotonic()<until:
|
|
time.sleep(.01)
|
|
assert (auth.settings.state_dir/'.credential.sock').exists()
|
|
with client(authz) as b:
|
|
r = b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC})
|
|
assert r.status_code == 202 and r.json()['accepted']
|
|
thread.join(timeout=3)
|
|
assert not thread.is_alive() and not errors
|
|
assert result[0]['credentials']['vault_password'] == SYNTHETIC
|
|
assert b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC}).status_code == 409
|
|
state = b.get('/api/v2/runs/'+ident+'/credential-status').json()
|
|
assert state['phase'] == 'claimed' and not state['can_submit']
|
|
assert SYNTHETIC.encode() not in auth.settings.database.read_bytes()
|
|
result[0]['credentials'].clear();result.clear()
|
|
|
|
|
|
def test_explicit_key_requirement_is_not_downgraded_when_vault_also_required(authz):
|
|
ident = make_job(authz, requirements=['vault_password', 'ssh_key_passphrase'])
|
|
with client(authz) as b:
|
|
r = b.get('/_partials/jobs/'+ident+'/credentials')
|
|
assert r.status_code == 200
|
|
assert 'data-key-source-choice' not in r.text
|
|
assert 'name="ssh_key_passphrase" maxlength="2048" required' in r.text
|
|
r = b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_credential_rate_limit_is_preserved_across_api_aliases(authz, monkeypatch):
|
|
ident = make_job(authz)
|
|
# This fixture isolates throttling; actual wire/claim is covered separately.
|
|
monkeypatch.setattr(service, 'submit', lambda *args: {'accepted': True})
|
|
with client(authz) as b:
|
|
for i in range(5):
|
|
route = '/api/v2/'+('runs' if i % 2 else 'jobs')+'/'+ident+'/credentials'
|
|
assert b.post(route, json={'vault_password': SYNTHETIC}).status_code == 202
|
|
assert b.post('/api/v2/runs/'+ident+'/credentials', json={'vault_password': SYNTHETIC}).status_code == 429
|