"""One-line bootstrap installer: token lifecycle and the two public downloads. The token is the whole authorization on a PC that has no agent, so these tests read the real routes rather than the helpers: that an admin mints and revokes, that anything not currently usable answers 404 rather than confirming itself, that only the package download spends a use, that the package can never carry a collector key, and that every fetch leaves an audit row with the caller's IP. """ import datetime import hashlib import io import json import zipfile import pytest from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import NameOID from plugins.endpoints.api import agentstore, siteagent from plugins.endpoints.models import EndpointBootstrapToken from shopdb.core.models import AuditLog, Setting TOKENS_URL = '/api/endpoints/siteagent/bootstrap/tokens' SCRIPT_URL = '/api/endpoints/siteagent/bootstrap.ps1' PACKAGE_URL = '/api/endpoints/siteagent/bootstrap/package' BASE_URL = 'https://shopdb.test/shopdb' def _make_cert(): key = ec.generate_private_key(ec.SECP256R1()) name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'server.test')]) now = datetime.datetime.now(datetime.timezone.utc) certificate = (x509.CertificateBuilder() .subject_name(name).issuer_name(name) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(now - datetime.timedelta(days=1)) .not_valid_after(now + datetime.timedelta(days=30)) .sign(key, hashes.SHA256())) pem = certificate.public_bytes(serialization.Encoding.PEM) spki = certificate.public_key().public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo) return pem, hashlib.sha256(spki).hexdigest() @pytest.fixture(autouse=True) def isolate(tmp_path, monkeypatch, app, db): """Empty agent store, no auto-detected certificate, a known base URL.""" store = tmp_path / 'siteagent' store.mkdir() monkeypatch.setattr(agentstore, '_siteagent_dir', lambda: str(store)) monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', ()) Setting.set('site_base_url', BASE_URL, valuetype='string') # The limiter is process-wide cache state; a noisy neighbour test would # otherwise leak its budget into this module. app.config['ENDPOINTS_BOOTSTRAP_RATELIMIT_ENABLED'] = False yield tmp_path app.config.pop('ENDPOINTS_BOOTSTRAP_RATELIMIT_ENABLED', None) @pytest.fixture def servercert(tmp_path, monkeypatch): pem, pin = _make_cert() path = tmp_path / 'server.crt' path.write_bytes(pem) monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', (str(path),)) return pin def _mint(client, auth_headers, **body): return client.post(TOKENS_URL, json=body or {}, headers=auth_headers) def _secret(client, auth_headers, **body): response = _mint(client, auth_headers, **body) assert response.status_code == 201, response.get_data(as_text=True) return response.get_json()['data'] def _files(response): assert response.status_code == 200, response.get_data(as_text=True)[:400] archive = zipfile.ZipFile(io.BytesIO(response.data)) return {name: archive.read(name) for name in archive.namelist()} # --------------------------------------------------------------------------- # Admin API # --------------------------------------------------------------------------- def test_generate_returns_the_secret_and_the_ready_made_command( client, db, auth_headers): data = _secret(client, auth_headers, hours=6, maxuses=3) secret = data['secret'] assert secret.startswith('sdbb_') assert len(secret) == len('sdbb_') + 43 assert data['command'] == ( 'powershell -ExecutionPolicy Bypass -c "irm ' '{0}/api/endpoints/siteagent/bootstrap.ps1?t={1} | iex"'.format( BASE_URL, secret)) token = data['token'] assert token['tokenprefix'] == secret[:12] assert token['maxuses'] == 3 assert token['usecount'] == 0 assert token['remaining'] == 3 assert token['state'] == 'active' assert token['createdby'] == 'testadmin' def test_defaults_are_a_day_and_twenty_five_uses(client, db, auth_headers): token = _secret(client, auth_headers)['token'] assert token['maxuses'] == 25 created = datetime.datetime.fromisoformat(token['createdat'][:-1]) expires = datetime.datetime.fromisoformat(token['expiresat'][:-1]) assert round((expires - created).total_seconds() / 3600) == 24 @pytest.mark.parametrize('body', [ {'hours': 0}, {'hours': 169}, {'maxuses': 0}, {'maxuses': 501}, {'hours': 'soon'}, ]) def test_out_of_range_inputs_are_refused(client, db, auth_headers, body): assert _mint(client, auth_headers, **body).status_code == 400 def test_list_never_carries_a_secret(client, db, auth_headers): data = _secret(client, auth_headers) response = client.get(TOKENS_URL, headers=auth_headers) assert response.status_code == 200 tokens = response.get_json()['data']['tokens'] assert len(tokens) == 1 assert tokens[0]['tokenprefix'] == data['secret'][:12] body = response.get_data(as_text=True) assert data['secret'] not in body assert 'secret' not in tokens[0] assert 'tokenhash' not in tokens[0] def test_revoke_kills_the_token(client, db, auth_headers): data = _secret(client, auth_headers) tokenid = data['token']['tokenid'] response = client.post('{0}/{1}/revoke'.format(TOKENS_URL, tokenid), headers=auth_headers) assert response.status_code == 200 assert response.get_json()['data']['token']['state'] == 'revoked' # Idempotent: revoking twice is not an error. assert client.post('{0}/{1}/revoke'.format(TOKENS_URL, tokenid), headers=auth_headers).status_code == 200 assert client.post('{0}/999/revoke'.format(TOKENS_URL), headers=auth_headers).status_code == 404 @pytest.mark.parametrize('method,path', [ ('post', TOKENS_URL), ('get', TOKENS_URL), ('post', TOKENS_URL + '/1/revoke'), ]) def test_only_an_admin_can_manage_tokens(client, db, member_headers, method, path): anonymous = getattr(client, method)(path) assert anonymous.status_code in (401, 422) member = getattr(client, method)(path, headers=member_headers) assert member.status_code == 403 def test_generate_and_revoke_write_audit_rows(client, db, auth_headers): data = _secret(client, auth_headers) client.post('{0}/{1}/revoke'.format(TOKENS_URL, data['token']['tokenid']), headers=auth_headers) rows = AuditLog.query.filter_by( entitytype='EndpointBootstrapToken').order_by(AuditLog.auditlogid).all() assert [row.action for row in rows] == ['created', 'updated'] assert all(row.entityname == data['secret'][:12] for row in rows) assert all(row.username == 'testadmin' for row in rows) # --------------------------------------------------------------------------- # bootstrap.ps1 # --------------------------------------------------------------------------- def test_script_is_served_for_a_good_token(client, db, auth_headers): secret = _secret(client, auth_headers)['secret'] response = client.get(SCRIPT_URL, query_string={'t': secret}) assert response.status_code == 200 assert response.mimetype == 'text/plain' script = response.get_data(as_text=True) script.encode('ascii') assert '{0}/api/endpoints/siteagent/bootstrap/package?t={1}'.format( BASE_URL, secret) in script assert 'Run as administrator' in script assert 'GE-Enforce' in script assert 'SHOPDB_BOOTSTRAP_FORCE' in script assert 'Settings > Endpoints > Enrollment' in script assert 'approved and enrolled' in script assert 'Get-FileHash' in script def test_script_bakes_the_staged_binary_hashes(client, db, auth_headers, tmp_path): payload = b'MZ' + b'agent-bytes' traypayload = b'MZ' + b'tray-bytes' agentstore.store_version('agent', '1.20.0', payload) agentstore.store_version('tray', '1.7.0', traypayload) secret = _secret(client, auth_headers)['secret'] script = client.get(SCRIPT_URL, query_string={'t': secret}).get_data(as_text=True) assert hashlib.sha256(payload).hexdigest() in script assert hashlib.sha256(traypayload).hexdigest() in script def test_script_fetch_does_not_spend_a_use(client, db, auth_headers): secret = _secret(client, auth_headers, maxuses=1)['secret'] for _ in range(3): assert client.get(SCRIPT_URL, query_string={'t': secret}).status_code == 200 row = EndpointBootstrapToken.query.one() assert row.usecount == 0 def test_script_fetch_is_audited_with_the_caller_ip(client, db, auth_headers): secret = _secret(client, auth_headers)['secret'] client.get(SCRIPT_URL, query_string={'t': secret}, environ_overrides={'REMOTE_ADDR': '10.0.0.9'}) row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken', action='requested').one() assert row.ipaddress == '10.0.0.9' assert '10.0.0.9' in row.details assert row.username is None # --------------------------------------------------------------------------- # The package # --------------------------------------------------------------------------- def test_package_is_served_and_spends_one_use(client, db, auth_headers, servercert): secret = _secret(client, auth_headers, maxuses=2)['secret'] files = _files(client.get(PACKAGE_URL, query_string={'t': secret}, environ_overrides={'REMOTE_ADDR': '10.0.0.8'})) assert 'install.ps1' in files assert 'appsettings.json' in files row = EndpointBootstrapToken.query.one() assert row.usecount == 1 assert row.lastusedip == '10.0.0.8' assert row.lastusedat is not None def test_package_carries_no_collector_key_and_does_carry_server_pins( client, db, auth_headers, servercert): secret = _secret(client, auth_headers)['secret'] files = _files(client.get(PACKAGE_URL, query_string={'t': secret})) for name, content in files.items(): assert b'CollectorKey' not in content, name script = files['install.ps1'].decode('ascii') assert '$sharedKeyMode = "omitted"' in script assert '[string[]] $ServerPins = @("{0}")'.format(servercert) in script settings = json.loads(files['appsettings.json']) assert settings['serverpins'] == [servercert] assert settings['apikey'] == '' assert b'Settings > Endpoints > Enrollment' in files['README.txt'] def test_package_stays_keyless_even_when_the_shared_key_is_allowed( client, db, auth_headers, servercert): Setting.set('plugin.endpoints.sharedkeyallowed', True, valuetype='boolean') secret = _secret(client, auth_headers)['secret'] files = _files(client.get(PACKAGE_URL, query_string={'t': secret})) assert b'CollectorKey' not in files['install.ps1'] def test_package_download_is_audited(client, db, auth_headers): secret = _secret(client, auth_headers)['secret'] client.get(PACKAGE_URL, query_string={'t': secret}, environ_overrides={'REMOTE_ADDR': '10.0.0.7'}) row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken', action='downloaded').one() assert row.ipaddress == '10.0.0.7' assert 'use 1 of' in row.details def test_a_token_is_refused_once_exhausted(client, db, auth_headers): secret = _secret(client, auth_headers, maxuses=1)['secret'] assert client.get(PACKAGE_URL, query_string={'t': secret}).status_code == 200 assert client.get(PACKAGE_URL, query_string={'t': secret}).status_code == 404 # The script route is gated on the same usability test. assert client.get(SCRIPT_URL, query_string={'t': secret}).status_code == 404 # --------------------------------------------------------------------------- # Everything unusable answers 404 # --------------------------------------------------------------------------- @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) @pytest.mark.parametrize('bad', ['', 'nonsense', 'sdbb_' + 'a' * 43, 'sdbb_short']) def test_an_unknown_token_is_not_found(client, db, url, bad): assert client.get(url, query_string={'t': bad}).status_code == 404 @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_no_token_at_all_is_not_found(client, db, url): assert client.get(url).status_code == 404 @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_an_expired_token_is_not_found(client, db, auth_headers, url): secret = _secret(client, auth_headers, hours=1)['secret'] row = EndpointBootstrapToken.query.one() row.expiresat = datetime.datetime.utcnow() - datetime.timedelta(minutes=1) db.session.commit() assert client.get(url, query_string={'t': secret}).status_code == 404 @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_a_revoked_token_is_not_found(client, db, auth_headers, url): data = _secret(client, auth_headers) client.post('{0}/{1}/revoke'.format(TOKENS_URL, data['token']['tokenid']), headers=auth_headers) assert client.get(url, query_string={'t': data['secret']}).status_code == 404 def test_a_bad_token_leaves_no_audit_row(client, db): client.get(SCRIPT_URL, query_string={'t': 'sdbb_' + 'a' * 43}) assert AuditLog.query.filter_by( entitytype='EndpointBootstrapToken').count() == 0 # --------------------------------------------------------------------------- # Rate limiting # --------------------------------------------------------------------------- def test_the_rate_limit_triggers_on_both_routes(client, db, app, auth_headers): secret = _secret(client, auth_headers, maxuses=500)['secret'] app.config['ENDPOINTS_BOOTSTRAP_RATELIMIT_ENABLED'] = True app.config['ENDPOINTS_BOOTSTRAP_RATELIMIT_MAX'] = 3 app.config['ENDPOINTS_BOOTSTRAP_RATELIMIT_WINDOW_SECONDS'] = 60 caller = {'REMOTE_ADDR': '10.9.9.9'} try: codes = [client.get(SCRIPT_URL, query_string={'t': secret}, environ_overrides=caller).status_code for _ in range(4)] assert codes == [200, 200, 200, 429] # The budget is per IP, not per route: the package route is already over. limited = client.get(PACKAGE_URL, query_string={'t': secret}, environ_overrides=caller) assert limited.status_code == 429 assert limited.get_json()['data']['error']['code'] == 'RATE_LIMITED' # A different caller is unaffected. assert client.get(SCRIPT_URL, query_string={'t': secret}, environ_overrides={'REMOTE_ADDR': '10.9.9.10'} ).status_code == 200 finally: app.config['ENDPOINTS_BOOTSTRAP_RATELIMIT_ENABLED'] = False app.config.pop('ENDPOINTS_BOOTSTRAP_RATELIMIT_MAX', None) app.config.pop('ENDPOINTS_BOOTSTRAP_RATELIMIT_WINDOW_SECONDS', None) # --------------------------------------------------------------------------- # Base URL resolution # --------------------------------------------------------------------------- def test_the_command_falls_back_to_the_request_when_no_site_url_is_set( client, db, auth_headers): Setting.set('site_base_url', '', valuetype='string') data = _secret(client, auth_headers) assert data['command'].startswith( 'powershell -ExecutionPolicy Bypass -c "irm https://localhost' '/api/endpoints/siteagent/bootstrap.ps1?t=')