Make rollback, first upgrades and pin changes work as tested on Windows
Found in an end-to-end test on a Windows 11 VM against a self-signed server.
- Rollback: a 1.20 agent re-enrolls within a second of an un-enroll, so an
older agent pushed afterwards arrived on an enrolled PC and was refused.
Revoke now takes {"blockreenroll": true, "reason": ...}; a blocked PC is
never issued a credential and keeps working on the shared key until
POST .../assets/<assetid>/allow-reenroll. The Enrollment page offers the
block when un-enrolling and shows blocked PCs with an allow button.
- First upgrade of old agents: agents before 1.20 check the server
certificate when downloading the upgrade, so a store wave failed on PCs that
do not trust a self-signed server while the Migration tab said ready. A
failed download now marks the PC blocked with "server certificate not
trusted by this PC". New Install-ShopDBServerCertificate.ps1 (GE-Enforce,
SYSTEM) adds a non-CA server certificate to the trusted roots after checking
its thumbprint; the server certificate page offers a .cer download and the
SHA-1 thumbprint.
- The generated installer writes ServerPins (with an optional next pin for a
planned rotation), can leave the shared key out so the PC waits for
approval, never touches an existing credential, and skips the trusted-root
import for a CA certificate.
- The agent re-reads server pins about every 60 seconds and recycles pooled
connections within 2 minutes, so a pin change needs no restart.
- An alert can only be acknowledged by a PC it targets.
- A wave no longer counts a PC as failed after a later wave upgraded it.
- The scripts report a pin mismatch only when the pin really is missing, and
otherwise log the real error.
- Runbook: trust step before the first wave, rollback order, pin re-read,
verifying the pin against the live URL; agent contract updated.
This commit is contained in:
cproudlock
@@ -267,6 +267,102 @@ def test_credential_is_invalid_after_revoke(client, db, app, keyheaders,
|
||||
assert reissued != secret
|
||||
|
||||
|
||||
def test_revoke_with_block_keeps_the_pc_off_credentials(client, db, app,
|
||||
keyheaders,
|
||||
auth_headers):
|
||||
from plugins.endpoints.models import EndpointCredential
|
||||
|
||||
assetid = _computer(db, 'BAY-01')
|
||||
secret = _credential(client, keyheaders)
|
||||
|
||||
revoke = client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
|
||||
json={'blockreenroll': True, 'reason': 'rollback'},
|
||||
headers=auth_headers)
|
||||
assert revoke.status_code == 200, revoke.get_data(as_text=True)
|
||||
data = revoke.get_json()['data']
|
||||
assert data['revokedat'] is not None
|
||||
assert data['reenrollblocked'] is True
|
||||
assert data['reenrollblockedreason'] == 'rollback'
|
||||
|
||||
# The current agent asks again at once: refused, with the reason.
|
||||
again = _enroll(client, keyheaders)
|
||||
assert again.status_code == 202, again.get_data(as_text=True)
|
||||
assert again.get_json()['data']['reason'] == 'reenroll blocked'
|
||||
assert client.get('/api/endpoints/alerts',
|
||||
headers={'X-Endpoint-Credential': secret}).status_code == 401
|
||||
|
||||
# An older agent on the shared key keeps working.
|
||||
assert client.get('/api/endpoints/alerts?hostname=BAY-01',
|
||||
headers=keyheaders).status_code == 200
|
||||
|
||||
listing = client.get('/api/endpoints/enrollment',
|
||||
headers=auth_headers).get_json()['data']
|
||||
row = next(r for r in listing['computers'] if r['assetid'] == assetid)
|
||||
assert row['reenrollblocked'] is True
|
||||
assert row['reenrollblockedreason'] == 'rollback'
|
||||
assert row['state'] == 'revoked'
|
||||
|
||||
# Blocked also means an approval cannot hand one out.
|
||||
requestid = again.get_json()['data']['requestid']
|
||||
approve = client.post(
|
||||
f'/api/endpoints/enrollment/requests/{requestid}/approve',
|
||||
json={'assetid': assetid}, headers=auth_headers)
|
||||
assert approve.status_code == 409
|
||||
|
||||
allow = client.post(
|
||||
f'/api/endpoints/enrollment/assets/{assetid}/allow-reenroll',
|
||||
headers=auth_headers)
|
||||
assert allow.status_code == 200, allow.get_data(as_text=True)
|
||||
assert allow.get_json()['data']['reenrollblocked'] is False
|
||||
row = EndpointCredential.query.filter_by(assetid=assetid).one()
|
||||
assert row.reenrollblockedreason is None
|
||||
|
||||
reissued = _credential(client, keyheaders)
|
||||
assert reissued != secret
|
||||
|
||||
|
||||
def test_blocked_pc_under_shared_key_off_is_refused_everything(
|
||||
client, db, app, keyheaders, auth_headers):
|
||||
assetid = _computer(db, 'BAY-01')
|
||||
_credential(client, keyheaders)
|
||||
client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
|
||||
json={'blockreenroll': True}, headers=auth_headers)
|
||||
_setting(db, 'plugin.endpoints.sharedkeyallowed', False)
|
||||
|
||||
refused = client.get('/api/endpoints/alerts?hostname=BAY-01',
|
||||
headers=keyheaders)
|
||||
assert refused.status_code == 401
|
||||
assert _code(refused) == 'SHARED_KEY_REFUSED'
|
||||
|
||||
|
||||
def test_revoke_block_validation_and_revoked_rows(client, db, app, keyheaders,
|
||||
auth_headers):
|
||||
assetid = _computer(db, 'BAY-01')
|
||||
nocredential = _computer(db, 'BAY-02', serialnumber='SER-2')
|
||||
url = f'/api/endpoints/enrollment/assets/{assetid}/revoke'
|
||||
_credential(client, keyheaders)
|
||||
|
||||
assert client.post(url, json={'blockreenroll': 'yes'},
|
||||
headers=auth_headers).status_code == 400
|
||||
assert client.post(url, json={'blockreenroll': True, 'reason': 'x' * 65},
|
||||
headers=auth_headers).status_code == 400
|
||||
|
||||
# Plain revoke, then block the already-revoked credential afterwards.
|
||||
assert client.post(url, headers=auth_headers).status_code == 200
|
||||
assert client.post(url, headers=auth_headers).status_code == 404
|
||||
blocked = client.post(url, json={'blockreenroll': True},
|
||||
headers=auth_headers)
|
||||
assert blocked.status_code == 200
|
||||
assert blocked.get_json()['data']['reenrollblocked'] is True
|
||||
|
||||
assert client.post(
|
||||
f'/api/endpoints/enrollment/assets/{nocredential}/revoke',
|
||||
json={'blockreenroll': True}, headers=auth_headers).status_code == 404
|
||||
assert client.post(
|
||||
f'/api/endpoints/enrollment/assets/{nocredential}/allow-reenroll',
|
||||
headers=auth_headers).status_code == 404
|
||||
|
||||
|
||||
def test_garbage_credential_is_invalid(client, db, app):
|
||||
response = client.get('/api/endpoints/commands/next',
|
||||
headers={'X-Endpoint-Credential': 'sdbe_nope'})
|
||||
@@ -349,7 +445,7 @@ def test_command_result_for_own_command_needs_no_hostname(client, db, app,
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
|
||||
|
||||
def _fleet_alert(db):
|
||||
def _fleet_alert(db, alertscope=None):
|
||||
from plugins.notifications.models import Notification, NotificationType
|
||||
|
||||
ntype = NotificationType(typename='Fleet Alert', typecolor='#d93f3f',
|
||||
@@ -358,7 +454,7 @@ def _fleet_alert(db):
|
||||
db.session.flush()
|
||||
note = Notification(notificationtypeid=ntype.notificationtypeid,
|
||||
notification='Line 3 is down', alertfleet=True,
|
||||
alerttimeout=60, isactive=True)
|
||||
alerttimeout=60, isactive=True, alertscope=alertscope)
|
||||
db.session.add(note)
|
||||
db.session.commit()
|
||||
return note.notificationid
|
||||
@@ -402,6 +498,37 @@ def test_alert_ack_with_no_assetid_means_the_caller(client, db, app,
|
||||
notificationid=notificationid, assetid=assetid).count() == 1
|
||||
|
||||
|
||||
def test_alert_ack_refuses_an_alert_not_scoped_to_the_caller(
|
||||
client, db, app, keyheaders):
|
||||
from plugins.notifications.models import NotificationAcknowledgement
|
||||
|
||||
_computer(db, 'BAY-01')
|
||||
_computer(db, 'BAY-02', serialnumber='SER-2')
|
||||
_computer(db, 'BAY-03', serialnumber='SER-3')
|
||||
secret = _credential(client, keyheaders)
|
||||
notificationid = _fleet_alert(db, alertscope={'hostname': ['BAY-02',
|
||||
'BAY-03']})
|
||||
url = f'/api/endpoints/alerts/{notificationid}/acknowledge'
|
||||
unknown = client.post('/api/endpoints/alerts/999999/acknowledge', json={},
|
||||
headers={'X-Endpoint-Credential': secret})
|
||||
|
||||
bycredential = client.post(url, json={},
|
||||
headers={'X-Endpoint-Credential': secret})
|
||||
assert bycredential.status_code == 404
|
||||
assert bycredential.get_json()['data']['error']['code'] == \
|
||||
unknown.get_json()['data']['error']['code']
|
||||
|
||||
bysharedkey = client.post(url, json={'hostname': 'BAY-03'},
|
||||
headers=keyheaders)
|
||||
assert bysharedkey.status_code == 201, bysharedkey.get_data(as_text=True)
|
||||
|
||||
_computer(db, 'BAY-04', serialnumber='SER-4')
|
||||
outside = client.post(url, json={'hostname': 'BAY-04'}, headers=keyheaders)
|
||||
assert outside.status_code == 404
|
||||
assert NotificationAcknowledgement.query.filter_by(
|
||||
notificationid=notificationid).count() == 1
|
||||
|
||||
|
||||
def test_collector_computers_refuses_another_hostname(client, db, app,
|
||||
keyheaders):
|
||||
_computer(db, 'BAY-01')
|
||||
@@ -487,6 +614,8 @@ def test_contract_exports_answer_through_the_endpoints_plugin(client, db, app,
|
||||
('post', '/api/endpoints/enrollment/requests/1/approve'),
|
||||
('post', '/api/endpoints/enrollment/requests/1/reject'),
|
||||
('post', '/api/endpoints/enrollment/assets/1/revoke'),
|
||||
('post', '/api/endpoints/enrollment/assets/1/allow-reenroll'),
|
||||
('get', '/api/endpoints/settings/server-cert/download'),
|
||||
('put', '/api/endpoints/enrollment/settings'),
|
||||
('post', '/api/endpoints/enrollment/migration/wave'),
|
||||
])
|
||||
@@ -586,4 +715,17 @@ def test_server_cert_reports_the_spki_pin(client, db, auth_headers, tmp_path,
|
||||
response = client.get('/api/endpoints/settings/server-cert',
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
assert response.get_json()['data']['spkisha256'] == expected
|
||||
data = response.get_json()['data']
|
||||
assert data['spkisha256'] == expected
|
||||
sha1 = certificate.fingerprint(hashes.SHA1()).hex().upper()
|
||||
assert data['sha1thumbprint'] == sha1
|
||||
assert ':' not in data['sha1thumbprint']
|
||||
|
||||
download = client.get('/api/endpoints/settings/server-cert/download',
|
||||
headers=auth_headers)
|
||||
assert download.status_code == 200
|
||||
assert download.mimetype == 'application/pkix-cert'
|
||||
assert 'attachment' in download.headers['Content-Disposition']
|
||||
assert '.cer' in download.headers['Content-Disposition']
|
||||
assert download.data == certificate.public_bytes(serialization.Encoding.DER)
|
||||
assert b'PRIVATE' not in download.data
|
||||
|
||||
@@ -86,7 +86,8 @@ def _assert_head_schema(connection):
|
||||
credentialcolumns = {c['name'] for c in inspector.get_columns('endpointcredentials')}
|
||||
assert credentialcolumns == {
|
||||
'credentialid', 'assetid', 'credentialhash', 'credentialprefix',
|
||||
'enrolledvia', 'issuedat', 'lastusedat', 'revokedat', 'revokedby'}
|
||||
'enrolledvia', 'issuedat', 'lastusedat', 'revokedat', 'revokedby',
|
||||
'reenrollblocked', 'reenrollblockedreason'}
|
||||
requestindexes = [tuple(ix['column_names']) for ix in
|
||||
inspector.get_indexes('endpointenrollrequests')]
|
||||
assert requestindexes.count(('hostname', 'serialnumber')) == 1
|
||||
|
||||
@@ -276,3 +276,140 @@ def test_wave_without_ready_pcs_creates_nothing(client, db, app, auth_headers):
|
||||
def test_wave_validation(client, db, app, auth_headers, body):
|
||||
assert client.post(WAVE_URL, json=body,
|
||||
headers=auth_headers).status_code == 400
|
||||
|
||||
|
||||
def test_a_reenroll_blocked_pc_is_blocked_not_ready(client, db, app,
|
||||
auth_headers):
|
||||
from plugins.endpoints.api.enrollment import issue_credential
|
||||
from plugins.endpoints.models import EndpointEnrollRequest
|
||||
|
||||
_publish_agent()
|
||||
assetid = _pc(db, 'PC-ROLLBACK')
|
||||
issue_credential(assetid, 'auto')
|
||||
db.session.commit()
|
||||
response = client.post(
|
||||
f'/api/endpoints/enrollment/assets/{assetid}/revoke',
|
||||
json={'blockreenroll': True, 'reason': 'rollback'},
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
# Its agent keeps asking; that request must not turn the row into pending.
|
||||
db.session.add(EndpointEnrollRequest(
|
||||
hostname='PC-ROLLBACK', serialnumber='SER', enrollnonce='n' * 40,
|
||||
status='pending', reason='reenroll blocked', matchedassetid=assetid,
|
||||
requestedat=_now(), lastseenat=_now()))
|
||||
db.session.commit()
|
||||
|
||||
row = _by_host(_rows(client, auth_headers))['PC-ROLLBACK']
|
||||
assert row['state'] == 'blocked'
|
||||
assert row['blockers'] == ['reenroll blocked']
|
||||
|
||||
wave = client.post(WAVE_URL, json={'assetids': [assetid]},
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert wave['skipped'] == [{'assetid': assetid,
|
||||
'reason': 'reenroll blocked'}]
|
||||
|
||||
client.post(f'/api/endpoints/enrollment/assets/{assetid}/allow-reenroll',
|
||||
headers=auth_headers)
|
||||
assert _by_host(_rows(client, auth_headers))['PC-ROLLBACK']['state'] \
|
||||
== 'pending'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('message,detail', [
|
||||
('Script download failed: The SSL connection could not be established, '
|
||||
'see inner exception.', None),
|
||||
('Script download failed', {'error': 'The remote certificate is invalid '
|
||||
'because of errors in the certificate chain: '
|
||||
'UntrustedRoot'}),
|
||||
('Invoke-WebRequest: Could not establish trust relationship for the '
|
||||
'SSL/TLS secure channel.', None),
|
||||
])
|
||||
def test_a_tls_trust_failure_blocks_the_pc(client, db, app, auth_headers,
|
||||
message, detail):
|
||||
import json
|
||||
|
||||
from plugins.endpoints.models import EndpointCommand
|
||||
|
||||
_publish_agent()
|
||||
assetid = _pc(db, 'PC-UNTRUSTED')
|
||||
otherid = _pc(db, 'PC-OTHERFAIL')
|
||||
agentregistry.register_current_agent_package()
|
||||
scriptid = agentupgradescript.seed_upgrade_script()['scriptid']
|
||||
db.session.add(EndpointCommand(
|
||||
assetid=assetid, action='run_script', target=f'script:{scriptid}',
|
||||
status=EndpointCommand.STATUS_FAILED, resultmessage=message,
|
||||
resultjson=json.dumps(detail) if detail is not None else None,
|
||||
completedat=_now()))
|
||||
db.session.add(EndpointCommand(
|
||||
assetid=otherid, action='run_script', target=f'script:{scriptid}',
|
||||
status=EndpointCommand.STATUS_FAILED,
|
||||
resultmessage='Script exited with code 1', completedat=_now()))
|
||||
db.session.commit()
|
||||
|
||||
rows = _by_host(_rows(client, auth_headers))
|
||||
assert rows['PC-UNTRUSTED']['state'] == 'blocked'
|
||||
assert rows['PC-UNTRUSTED']['blockers'] == [
|
||||
'server certificate not trusted by this PC']
|
||||
assert rows['PC-OTHERFAIL']['state'] == 'ready'
|
||||
|
||||
|
||||
def test_a_later_success_clears_the_tls_blocker(client, db, app, auth_headers):
|
||||
from plugins.endpoints.models import EndpointCommand
|
||||
|
||||
_publish_agent()
|
||||
assetid = _pc(db, 'PC-FIXED')
|
||||
agentregistry.register_current_agent_package()
|
||||
scriptid = agentupgradescript.seed_upgrade_script()['scriptid']
|
||||
db.session.add(EndpointCommand(
|
||||
assetid=assetid, action='run_script', target=f'script:{scriptid}',
|
||||
status=EndpointCommand.STATUS_FAILED,
|
||||
resultmessage='RemoteCertificateChainErrors', completedat=_now()))
|
||||
db.session.flush()
|
||||
db.session.add(EndpointCommand(
|
||||
assetid=assetid, action='run_script', target='script:999',
|
||||
status=EndpointCommand.STATUS_SUCCEEDED, completedat=_now()))
|
||||
db.session.commit()
|
||||
|
||||
assert _by_host(_rows(client, auth_headers))['PC-FIXED']['state'] == 'ready'
|
||||
|
||||
|
||||
def test_wave_failed_forgets_a_pc_a_later_wave_upgraded(client, db, app,
|
||||
auth_headers):
|
||||
from plugins.endpoints.models import EndpointCommand, EndpointState
|
||||
|
||||
_publish_agent()
|
||||
upgraded = _pc(db, 'PC-UPGRADED')
|
||||
retried = _pc(db, 'PC-RETRIED')
|
||||
stuck = _pc(db, 'PC-STUCK')
|
||||
first = client.post(WAVE_URL, json={'assetids': [upgraded, retried, stuck]},
|
||||
headers=auth_headers)
|
||||
assert first.status_code == 201, first.get_data(as_text=True)
|
||||
for command in EndpointCommand.query.all():
|
||||
command.status = EndpointCommand.STATUS_FAILED
|
||||
db.session.commit()
|
||||
assert _rows(client, auth_headers)['waves'][0]['failed'] == 3
|
||||
|
||||
second = client.post(WAVE_URL, json={'assetids': [upgraded, retried]},
|
||||
headers=auth_headers)
|
||||
assert second.status_code == 201, second.get_data(as_text=True)
|
||||
# The later wave replaced the agent on PC-UPGRADED, which restarted before
|
||||
# reporting, so its command expired; it has not enrolled yet. PC-RETRIED's
|
||||
# push reported success and is waiting to enroll on its old version string.
|
||||
latest = {command.assetid: command for command in EndpointCommand.query
|
||||
.filter_by(status=EndpointCommand.STATUS_PENDING).all()}
|
||||
latest[upgraded].status = EndpointCommand.STATUS_EXPIRED
|
||||
latest[retried].status = EndpointCommand.STATUS_SUCCEEDED
|
||||
EndpointState.query.filter_by(assetid=upgraded).one().agentversion = '1.20.0'
|
||||
db.session.commit()
|
||||
|
||||
failed = {wave['waveid']: wave['failed']
|
||||
for wave in _rows(client, auth_headers)['waves']}
|
||||
assert failed == {first.get_json()['data']['waveid']: 1,
|
||||
second.get_json()['data']['waveid']: 0}
|
||||
|
||||
# A newer failure counts again, in every wave the PC belongs to.
|
||||
latest[retried].status = EndpointCommand.STATUS_FAILED
|
||||
db.session.commit()
|
||||
failed = {wave['waveid']: wave['failed']
|
||||
for wave in _rows(client, auth_headers)['waves']}
|
||||
assert failed == {first.get_json()['data']['waveid']: 2,
|
||||
second.get_json()['data']['waveid']: 1}
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
"""Site agent package build: server pins, optional shared key, Root import gate.
|
||||
|
||||
The package is the only place a PC gets its first trust anchor and (optionally)
|
||||
the shared fleet key, so these tests read the actual ZIP the route returns
|
||||
rather than the helper in isolation.
|
||||
"""
|
||||
|
||||
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 shopdb.core.models.setting import Setting
|
||||
|
||||
PACKAGE_URL = '/api/endpoints/siteagent/package'
|
||||
STATUS_URL = '/api/endpoints/siteagent'
|
||||
SECRET = 'sharedkeyvalue-0123456789abcdef'
|
||||
|
||||
|
||||
def _make_cert(isca):
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,
|
||||
'ca.test' if isca else 'server.test')])
|
||||
now = datetime.datetime.now(datetime.timezone.utc)
|
||||
builder = (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)))
|
||||
if isca:
|
||||
builder = builder.add_extension(
|
||||
x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
||||
certificate = builder.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):
|
||||
"""Empty agent store and no auto-detected server certificate by default."""
|
||||
store = tmp_path / 'siteagent'
|
||||
store.mkdir()
|
||||
monkeypatch.setattr(agentstore, '_siteagent_dir', lambda: str(store))
|
||||
monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', ())
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def servercert(tmp_path, monkeypatch):
|
||||
"""A self-signed leaf staged as this server's own certificate."""
|
||||
pem, pin = _make_cert(isca=False)
|
||||
path = tmp_path / 'server.crt'
|
||||
path.write_bytes(pem)
|
||||
monkeypatch.setattr(siteagent, '_CERT_CANDIDATES', (str(path),))
|
||||
return pin
|
||||
|
||||
|
||||
def _build(client, headers, **fields):
|
||||
body = {'baseurl': 'https://shopdb.test', 'cadenceseconds': 900}
|
||||
body.update(fields)
|
||||
return client.post(PACKAGE_URL, json=body, headers=headers)
|
||||
|
||||
|
||||
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()}
|
||||
|
||||
|
||||
def test_installer_writes_current_server_pin(client, db, auth_headers, servercert):
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET,
|
||||
includesharedkey=True, bundlecert=True))
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
|
||||
assert '[string[]] $ServerPins = @("{0}")'.format(servercert) in script
|
||||
assert 'Set-HklmMultiStringBothViews -SubKey "SOFTWARE\\GE\\ShopDB" -Name "ServerPins"' in script
|
||||
assert '[Microsoft.Win32.RegistryValueKind]::MultiString' in script
|
||||
assert '-ceq $wanted' in script
|
||||
settings = json.loads(files['appsettings.json'])
|
||||
assert settings['serverpins'] == [servercert]
|
||||
|
||||
|
||||
def test_pin_comes_from_server_cert_even_when_not_bundled(
|
||||
client, db, auth_headers, servercert):
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET))
|
||||
assert 'server.crt' not in files
|
||||
assert json.loads(files['appsettings.json'])['serverpins'] == [servercert]
|
||||
|
||||
|
||||
def test_next_pin_is_shipped_beside_current(client, db, auth_headers, servercert):
|
||||
nextpin = 'ab' * 32
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET,
|
||||
nextserverpin=nextpin))
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
|
||||
assert '@("{0}", "{1}")'.format(servercert, nextpin) in script
|
||||
assert json.loads(files['appsettings.json'])['serverpins'] == [servercert, nextpin]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('badpin', ['AB' * 32, 'ab' * 31, 'zz' * 32])
|
||||
def test_next_pin_must_be_64_lowercase_hex(client, db, auth_headers, servercert, badpin):
|
||||
response = _build(client, auth_headers, collectorkey=SECRET, nextserverpin=badpin)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_next_pin_without_a_current_pin_is_refused(client, db, auth_headers):
|
||||
response = _build(client, auth_headers, collectorkey=SECRET,
|
||||
nextserverpin='ab' * 32)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_shared_key_on_keeps_key_provisioning(client, db, auth_headers, servercert):
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET,
|
||||
includesharedkey='true'))
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
|
||||
assert '[string] $CollectorKey = "{0}"'.format(SECRET) in script
|
||||
assert '-Name "CollectorKey" -Value $CollectorKey' in script
|
||||
assert 'Settings > Endpoints > Enrollment' not in script
|
||||
|
||||
|
||||
def test_shared_key_off_leaves_no_key_anywhere(client, db, auth_headers, servercert):
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET,
|
||||
includesharedkey=False))
|
||||
|
||||
for name, content in files.items():
|
||||
assert b'CollectorKey' not in content, name
|
||||
assert SECRET.encode('ascii') not in content, name
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
assert 'Settings > Endpoints > Enrollment' in script
|
||||
assert '$sharedKeyMode = "omitted"' in script
|
||||
assert json.loads(files['appsettings.json'])['apikey'] == ''
|
||||
assert b'Settings > Endpoints > Enrollment' in files['README.txt']
|
||||
|
||||
|
||||
def test_shared_key_default_follows_site_setting(client, db, auth_headers, servercert):
|
||||
# Setting on (default): key required.
|
||||
assert _build(client, auth_headers).status_code == 400
|
||||
|
||||
Setting.set('plugin.endpoints.sharedkeyallowed', False, valuetype='boolean')
|
||||
status = client.get(STATUS_URL, headers=auth_headers).get_json()['data']
|
||||
assert status['sharedkeyallowed'] is False
|
||||
assert status['serverpin'] == servercert
|
||||
|
||||
files = _files(_build(client, auth_headers, collectorkey=SECRET))
|
||||
assert b'CollectorKey' not in files['install.ps1']
|
||||
assert SECRET.encode('ascii') not in files['install.ps1']
|
||||
|
||||
|
||||
def test_installer_never_removes_enrolled_credential(client, db, auth_headers, servercert):
|
||||
script = _files(_build(client, auth_headers, collectorkey=SECRET))[
|
||||
'install.ps1'].decode('ascii')
|
||||
assert 'existing enrolled credential found; kept as is.' in script
|
||||
for line in script.splitlines():
|
||||
if 'Remove-Item' in line:
|
||||
assert 'credential' not in line.lower(), line
|
||||
|
||||
|
||||
def test_ca_certificate_is_not_imported_into_root(client, db, auth_headers, servercert):
|
||||
capem, capin = _make_cert(isca=True)
|
||||
response = client.post(
|
||||
PACKAGE_URL,
|
||||
data={'baseurl': 'https://shopdb.test', 'collectorkey': SECRET,
|
||||
'cadenceseconds': '900',
|
||||
'certfile': (io.BytesIO(capem), 'ca.crt')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
files = _files(response)
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
|
||||
assert 'server.crt' in files
|
||||
assert '$trustCertInRoot = $false' in script
|
||||
assert 'is a CA certificate; not importing it' in script
|
||||
# A CA cannot be the pinned leaf; the server's own leaf is pinned instead.
|
||||
pins = json.loads(files['appsettings.json'])['serverpins']
|
||||
assert pins == [servercert]
|
||||
assert capin not in script
|
||||
|
||||
|
||||
def test_leaf_certificate_is_imported_into_root(client, db, auth_headers):
|
||||
leafpem, leafpin = _make_cert(isca=False)
|
||||
response = client.post(
|
||||
PACKAGE_URL,
|
||||
data={'baseurl': 'https://shopdb.test', 'collectorkey': SECRET,
|
||||
'cadenceseconds': '900',
|
||||
'certfile': (io.BytesIO(leafpem), 'server.crt')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
files = _files(response)
|
||||
script = files['install.ps1'].decode('ascii')
|
||||
|
||||
assert '$trustCertInRoot = $true' in script
|
||||
assert 'Import-Certificate -FilePath $certPath -CertStoreLocation Cert:\\LocalMachine\\Root' in script
|
||||
assert json.loads(files['appsettings.json'])['serverpins'] == [leafpin]
|
||||
|
||||
|
||||
def test_install_script_helper_without_pins_or_key():
|
||||
script = siteagent._install_script('https://shopdb.test', '', 900,
|
||||
serverpins=[], includesharedkey=False)
|
||||
assert '[string[]] $ServerPins = @()' in script
|
||||
assert 'CollectorKey' not in script
|
||||
assert 'ServerPins left as is' in script
|
||||
|
||||
Reference in New Issue
Block a user