ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2f98188feb1fddb9dd0c30afb13b9b239b926e47
shopdb-flask/tests/test_plugins/test_endpoints_bootstrap_install.py
cproudlock 987f5c0d0f Stop the one-line install handing a standard user a way in 
The bootstrap command staged its script into C:\Windows\Temp and then ran it
elevated. That directory grants Users create-file with CREATOR OWNER full
control, so somebody without admin rights can pre-plant the exact path the
elevated run is about to execute. It now stages under Program Files, creates the
directory with an explicit DACL when it is missing, and refuses to run a staged
file that is not owned by an admin principal. A failed hardening is loud rather
than silent: the run stops instead of carrying on with a file it could not
protect.

The token moved out of the query string into a header. In a query string it
lands verbatim in the reverse proxy and IIS logs, which is a credential written
to a file nobody thinks of as a secret store. The query form is still accepted,
because bays provisioned before this change are holding install commands that
use it, and it logs that it was used - without logging the token - so a site can
see who still needs a re-issued command before turning it off with
plugin.endpoints.bootstrapquerytoken.

Two smaller things this also closes:

A request with no token at all answered 400 naming the header, which told an
anonymous caller the state of that switch and contradicted the module's own
promise that anything unusable answers 404. No token is now 404. The 400 is kept
for a token genuinely presented in the query form while the query form is off,
which is a caller who needs to know their command is stale.

The rollout documentation was republishing the vulnerable one-liner verbatim, so
a tech following the runbook reinstated exactly what this fixes. Both documents
now carry the current command.

Also corrects the operator-facing copy for the token, which named a query
argument and a header that do not exist: it is ?t= and X-Endpoint-Bootstrap-Token,
per bootstrap.py's own constants. Anybody following the old text got a 404 with
no reason given.

Both switches added in this range are now declared with the other plugin
settings and reachable from their settings page. Read through Setting.get but
never declared, they were rows that did not exist until someone wrote one by
hand, which on an air-gapped box means the feature was unreachable.
2026-09-18 13:30:08 -04:00

678 lines
30 KiB
Python
Raw Blame History

"""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)
Reference in New Issue View Git Blame Copy Permalink