"""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 logging 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' TOKEN_HEADER = 'X-Endpoint-Bootstrap-Token' QUERY_SETTING = 'plugin.endpoints.bootstrapquerytoken' # Admin-only staging, NOT C:\Windows\Temp: a standard user can pre-plant a # path there and have it run elevated. STAGE_DIR = r'C:\Program Files\ShopDB\bootstrap' STAGED_SCRIPT = STAGE_DIR + r'\shopdb-bootstrap.ps1' ADMIN_SID = 'S-1-5-32-544' SYSTEM_SID = 'S-1-5-18' # The marker the admin "Download .ps1" link carries, so the server can tell an # admin's browser click apart from a bay still on a pre-change pasted command. UI_LINK_QUERY = 'via=download' def _expected_command(secret, base=BASE_URL): """Stage admin-only, download with the token in a header, check the owner, run, delete. Never `irm | iex` - Defender kills that shape.""" return ( 'powershell -ExecutionPolicy Bypass -c "' "try {{ New-Item -ItemType Directory -Force -Path '{dir}' " '| Out-Null; ' "icacls '{dir}' /inheritance:r /grant '*{admin}:(OI)(CI)F' " "'*{system}:(OI)(CI)F' | Out-Null; " 'if ((Get-Variable LASTEXITCODE -ValueOnly) -ne 0) ' "{{ Write-Host 'Refusing to run: could not restrict " "{dir} to administrators (icacls failed). Fix the directory " "or delete it, then run the command again.' -ForegroundColor Red; " 'exit 6 }}; ' "Remove-Item '{path}' -Force -ErrorAction SilentlyContinue; " "Invoke-WebRequest -UseBasicParsing " "'{base}/api/endpoints/siteagent/bootstrap.ps1' " "-Headers @{{'{header}'='{secret}'}} -OutFile '{path}'; " "if ((Get-Acl '{path}').GetOwner(" '[Security.Principal.SecurityIdentifier]).Value -notin ' "@('{admin}','{system}'," '[Security.Principal.WindowsIdentity]::GetCurrent().User.Value)) ' "{{ Write-Host 'Refusing to run: the staged bootstrap script is not " "owned by an administrator.' -ForegroundColor Red; exit 5 }}; " "& '{path}' }} finally {{ Remove-Item '{path}' -Force " '-ErrorAction SilentlyContinue }}"' ).format(dir=STAGE_DIR, path=STAGED_SCRIPT, base=base, header=TOKEN_HEADER, secret=secret, admin=ADMIN_SID, system=SYSTEM_SID) def _token_header(secret): return {TOKEN_HEADER: secret} 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'] == _expected_command(secret) assert data['scripturl'] == ( '{0}/api/endpoints/siteagent/bootstrap.ps1?t={1}&{2}'.format( BASE_URL, secret, UI_LINK_QUERY)) 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_the_command_downloads_and_runs_instead_of_piping_into_iex( client, db, auth_headers): """Defender detects `irm ... | iex` as Trojan:Win32/Commando.A!ml and kills the process before a line of it runs, so the published command downloads to a file, runs it and removes it. It also stays one pasteable line, and holds no `$`: pasted into a PowerShell prompt, the parent would expand those before the child process ever saw them.""" data = _secret(client, auth_headers) command = data['command'] assert ' | iex' not in command assert 'iex' not in command.lower() assert '$' not in command assert '\n' not in command assert data['secret'] in command assert "Remove-Item '{0}' -Force".format(STAGED_SCRIPT) in command assert 'finally' in command # The link the UI offers instead is the route that really serves it. assert client.get(data['scripturl'].split(BASE_URL, 1)[1] ).status_code == 200 def test_the_command_stages_where_only_administrators_can_write( client, db, auth_headers): """C:\\Windows\\Temp lets any standard user create the exact path the command runs elevated, so the staging directory moved under Program Files, is created with an admin-only DACL, and the command refuses to run a staged file owned by anyone but an administrator.""" command = _secret(client, auth_headers)['command'] assert r'C:\Windows\Temp' not in command assert "-OutFile '{0}'".format(STAGED_SCRIPT) in command assert "New-Item -ItemType Directory -Force -Path '{0}'".format( STAGE_DIR) in command assert "icacls '{0}' /inheritance:r".format(STAGE_DIR) in command # Well-known SIDs, so a renamed or localized group cannot widen the grant. assert "/grant '*{0}:(OI)(CI)F' '*{1}:(OI)(CI)F'".format( ADMIN_SID, SYSTEM_SID) in command assert "Get-Acl '{0}'".format(STAGED_SCRIPT) in command assert 'GetOwner([Security.Principal.SecurityIdentifier]).Value' in command assert "-notin @('{0}','{1}',".format(ADMIN_SID, SYSTEM_SID) in command assert '[Security.Principal.WindowsIdentity]::GetCurrent().User.Value' \ in command assert 'not owned by an administrator' in command assert 'exit 5' in command # A leftover from an earlier run is cleared before the download, so the # owner test cannot pass on stale content. assert command.index("Remove-Item '{0}'".format(STAGED_SCRIPT)) < \ command.index('Invoke-WebRequest') def test_the_command_sends_the_token_in_a_header_not_the_query_string( client, db, auth_headers): """A query string lands verbatim in the nginx and gunicorn access logs.""" data = _secret(client, auth_headers) command = data['command'] assert '?t=' not in command assert "-Headers @{{'{0}'='{1}'}}".format(TOKEN_HEADER, data['secret']) in command assert "'{0}/api/endpoints/siteagent/bootstrap.ps1'".format( BASE_URL) in command # The browser download link is the one place that cannot send a header, # and it is marked so its own fetches are not counted as un-migrated bays. assert data['scripturl'].endswith( '?t={0}&{1}'.format(data['secret'], UI_LINK_QUERY)) def test_a_failed_acl_hardening_stops_the_command(client, db, auth_headers): """icacls is a native command: it raises nothing and $ErrorActionPreference never applies to it, so an unchecked call leaves a staging directory that still inherits the parent DACL and the line downloads into it anyway. The exit code is read sigil-free, because the whole line must hold no `$`.""" command = _secret(client, auth_headers)['command'] assert '(Get-Variable LASTEXITCODE -ValueOnly) -ne 0' in command assert 'LASTEXITCODE' in command and '$LASTEXITCODE' not in command assert 'icacls failed' in command assert 'exit 6' in command # Checked before anything is fetched into the directory, not after. assert command.index('LASTEXITCODE') > command.index('icacls') assert command.index('LASTEXITCODE') < command.index('Invoke-WebRequest') 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"'.format( BASE_URL) 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_the_generated_script_fetches_the_package_with_the_header( client, db, auth_headers): """Same reason as the one-liner: the package URL must not carry the token, or every agent install writes it into an access log.""" secret = _secret(client, auth_headers)['secret'] script = client.get(SCRIPT_URL, headers=_token_header(secret)).get_data(as_text=True) assert '?t=' not in script assert '$BootstrapPackageHeaders = @{{ "{0}" = "{1}" }}'.format( TOKEN_HEADER, secret) in script assert ('Invoke-WebRequest -Uri $BootstrapPackageUrl ' '-Headers $BootstrapPackageHeaders') in script def test_both_routes_take_the_token_in_a_header(client, db, auth_headers, servercert): secret = _secret(client, auth_headers, maxuses=2)['secret'] script = client.get(SCRIPT_URL, headers=_token_header(secret)) assert script.status_code == 200 assert 'ShopDB endpoint agent bootstrap' in script.get_data(as_text=True) assert 'install.ps1' in _files( client.get(PACKAGE_URL, headers=_token_header(secret))) assert EndpointBootstrapToken.query.one().usecount == 1 @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_a_bad_header_token_is_not_found(client, db, auth_headers, url): """The header is read first, so a good ?t= cannot rescue a bad header - one request carries one token.""" secret = _secret(client, auth_headers)['secret'] response = client.get(url, query_string={'t': secret}, headers=_token_header('sdbb_' + 'a' * 43)) assert response.status_code == 404 def test_the_query_form_still_works_and_is_audited_as_deprecated( client, db, auth_headers, caplog): """A bay running a command pasted before this change must keep working, so the query form stays accepted; it is flagged in the audit row and warned about in the log, with no token value in either.""" secret = _secret(client, auth_headers)['secret'] with caplog.at_level(logging.WARNING): assert client.get(SCRIPT_URL, query_string={'t': secret}, environ_overrides={'REMOTE_ADDR': '10.0.0.6'} ).status_code == 200 row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken', action='requested').one() assert 'query-string token, deprecated' in row.details assert secret not in row.details warnings = [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] assert any('query string' in message and '10.0.0.6' in message for message in warnings) assert all(secret not in message for message in warnings) def test_the_header_form_is_not_flagged_as_deprecated(client, db, auth_headers, caplog): secret = _secret(client, auth_headers)['secret'] with caplog.at_level(logging.WARNING): client.get(SCRIPT_URL, headers=_token_header(secret)) row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken', action='requested').one() assert 'deprecated' not in row.details assert not [record for record in caplog.records if record.levelno >= logging.WARNING] def test_the_admin_download_link_is_not_counted_as_an_un_migrated_bay( client, db, auth_headers, caplog): """The rollout finds un-migrated bays by the deprecation WARNING, and the UI Download .ps1 link has to use the query form. Marked, it logs at info and is audited as the admin path, so an admin testing a token does not invent a bay that is not there.""" data = _secret(client, auth_headers) path = data['scripturl'].split(BASE_URL, 1)[1] with caplog.at_level(logging.INFO): assert client.get(path, environ_overrides={'REMOTE_ADDR': '10.0.0.7'} ).status_code == 200 row = AuditLog.query.filter_by(entitytype='EndpointBootstrapToken', action='requested').one() assert 'admin download link' in row.details assert 'deprecated' not in row.details assert data['secret'] not in row.details assert not [record for record in caplog.records if record.levelno >= logging.WARNING] infos = [record.getMessage() for record in caplog.records if record.levelno == logging.INFO] assert any('Download .ps1' in message and '10.0.0.7' in message for message in infos) assert all(data['secret'] not in message for message in infos) def test_no_log_or_audit_row_ever_carries_the_token_value( client, db, auth_headers, caplog, servercert): secret = _secret(client, auth_headers, maxuses=4)['secret'] with caplog.at_level(logging.DEBUG): client.get(SCRIPT_URL, query_string={'t': secret}) client.get(SCRIPT_URL, headers=_token_header(secret)) client.get(PACKAGE_URL, query_string={'t': secret}) client.get(PACKAGE_URL, headers=_token_header(secret)) assert all(secret not in record.getMessage() for record in caplog.records) rows = AuditLog.query.filter_by( entitytype='EndpointBootstrapToken').all() assert rows for row in rows: assert secret not in (row.details or '') # The 12-char display prefix is the identifier the UI already shows. assert row.entityname == secret[:12] @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_the_query_form_can_be_turned_off_once_a_site_has_re_issued( client, db, auth_headers, url): """Off is the hardened end state, not the default: the UI download link uses the query form too, so a site only turns it off deliberately.""" secret = _secret(client, auth_headers)['secret'] Setting.set(QUERY_SETTING, False, valuetype='boolean') refused = client.get(url, query_string={'t': secret}) assert refused.status_code == 400 assert TOKEN_HEADER in refused.get_json()['data']['error']['message'] # The admin Download .ps1 link is a query token too: the setting is about # a live token landing in an access log, and that link puts one there. assert client.get(url, query_string={'t': secret, 'via': 'download'} ).status_code == 400 # Refused before any lookup, so it confirms nothing about the token. assert EndpointBootstrapToken.query.one().usecount == 0 assert AuditLog.query.filter_by( entitytype='EndpointBootstrapToken', action='requested').count() == 0 assert client.get(url, headers=_token_header(secret)).status_code == 200 @pytest.mark.parametrize('url', [SCRIPT_URL, PACKAGE_URL]) def test_no_token_is_still_not_found_when_the_query_form_is_off( client, db, auth_headers, url): """A bare GET presented no token, so it is unusable and answers 404 like every other unusable request. Calling it a query token answered 400 naming the header, which handed an anonymous caller the state of the switch.""" _secret(client, auth_headers) Setting.set(QUERY_SETTING, False, valuetype='boolean') assert client.get(url).status_code == 404 # An empty ?t= presented nothing either. assert client.get(url, query_string={'t': ''}).status_code == 404 # And it is the same answer with the query form still on. Setting.set(QUERY_SETTING, True, valuetype='boolean') assert client.get(url).status_code == 404 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'] == _expected_command(data['secret'], base='https://localhost') assert data['scripturl'].startswith( 'https://localhost/api/endpoints/siteagent/bootstrap.ps1?t=') assert data['scripturl'].endswith('&' + UI_LINK_QUERY)