da4b224ac1
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.
732 lines
30 KiB
Python
732 lines
30 KiB
Python
"""Per-PC endpoint credentials: enrollment, authentication and identity binding.
|
|
|
|
A PC enrolls once on the shared fleet key and gets its own secret. From then on
|
|
the server derives the calling asset from that secret, so a request naming a
|
|
different PC is refused (ASSET_MISMATCH), the shared key is refused for that PC
|
|
(SHARED_KEY_REFUSED), and a revoked secret is refused outright
|
|
(CREDENTIAL_INVALID) so the agent knows to enroll again.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from shopdb.core.models import Asset, AssetType, Setting
|
|
|
|
|
|
COLLECTOR_KEY = 'testcollectorkey'
|
|
ENROLL_URL = '/api/endpoints/enroll'
|
|
NONCE = 'n' * 40
|
|
OTHER_NONCE = 'm' * 40
|
|
|
|
|
|
@pytest.fixture
|
|
def keyheaders(app):
|
|
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
|
return {'X-API-Key': COLLECTOR_KEY}
|
|
|
|
|
|
def _computer(db, hostname, serialnumber='SER-1'):
|
|
from plugins.computers.models import Computer
|
|
|
|
atype = AssetType.query.filter_by(assettype='computer').first()
|
|
if atype is None:
|
|
atype = AssetType(assettype='computer')
|
|
db.session.add(atype)
|
|
db.session.flush()
|
|
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
|
serialnumber=serialnumber, isactive=True)
|
|
db.session.add(asset)
|
|
db.session.flush()
|
|
db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
|
|
db.session.commit()
|
|
return asset.assetid
|
|
|
|
|
|
def _enroll_body(hostname='BAY-01', serialnumber='SER-1', nonce=NONCE):
|
|
return {'hostname': hostname, 'serialnumber': serialnumber,
|
|
'deviceid': None, 'enrollnonce': nonce, 'agentversion': '1.20.0'}
|
|
|
|
|
|
def _enroll(client, headers, **kwargs):
|
|
return client.post(ENROLL_URL, json=_enroll_body(**kwargs), headers=headers)
|
|
|
|
|
|
def _credential(client, keyheaders, **kwargs):
|
|
response = _enroll(client, keyheaders, **kwargs)
|
|
assert response.status_code == 201, response.get_data(as_text=True)
|
|
return response.get_json()['data']['credential']
|
|
|
|
|
|
def _setting(db, key, value):
|
|
Setting.set(key, value, valuetype='boolean', category='plugin.endpoints')
|
|
db.session.commit()
|
|
|
|
|
|
def _code(response):
|
|
return response.get_json()['data']['error']['code']
|
|
|
|
|
|
# -- enroll ------------------------------------------------------------------
|
|
|
|
def test_auto_enroll_issues_a_credential_once(client, db, app, keyheaders):
|
|
from plugins.endpoints.models import EndpointCredential
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
response = _enroll(client, keyheaders)
|
|
|
|
assert response.status_code == 201, response.get_data(as_text=True)
|
|
data = response.get_json()['data']
|
|
assert data['status'] == 'enrolled'
|
|
assert data['assetid'] == assetid
|
|
assert data['credential'].startswith('sdbe_')
|
|
assert len(data['credential']) == 48
|
|
|
|
row = EndpointCredential.query.filter_by(assetid=assetid).one()
|
|
assert row.enrolledvia == 'auto'
|
|
assert row.credentialprefix == data['credential'][:12]
|
|
# Hash only, never the secret.
|
|
assert data['credential'] not in (row.credentialhash, row.credentialprefix)
|
|
|
|
again = _enroll(client, keyheaders)
|
|
assert again.status_code == 202
|
|
assert again.get_json()['data']['reason'] == 'already enrolled'
|
|
|
|
|
|
@pytest.mark.parametrize('setup,body,reason', [
|
|
('none', {}, 'unknown hostname'),
|
|
('duplicate', {}, 'duplicate hostname'),
|
|
('noserial', {}, 'no serial on record'),
|
|
('serial', {'serialnumber': 'OTHER'}, 'serial mismatch'),
|
|
('noauto', {}, 'awaiting approval'),
|
|
])
|
|
def test_each_pending_reason(client, db, app, keyheaders, setup, body, reason):
|
|
if setup == 'duplicate':
|
|
_computer(db, 'BAY-01')
|
|
_computer(db, 'bay-01')
|
|
elif setup == 'noserial':
|
|
_computer(db, 'BAY-01', serialnumber=None)
|
|
elif setup in ('serial', 'noauto'):
|
|
_computer(db, 'BAY-01')
|
|
if setup == 'noauto':
|
|
_setting(db, 'plugin.endpoints.autoenroll', False)
|
|
|
|
response = _enroll(client, keyheaders, **body)
|
|
|
|
assert response.status_code == 202, response.get_data(as_text=True)
|
|
data = response.get_json()['data']
|
|
assert data['status'] == 'pending'
|
|
assert data['reason'] == reason
|
|
assert data['requestid']
|
|
|
|
|
|
def test_serial_compare_ignores_case_and_whitespace(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01', serialnumber=' ser-1 ')
|
|
assert _enroll(client, keyheaders, serialnumber='SER-1').status_code == 201
|
|
|
|
|
|
def test_enroll_validates_the_nonce(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
for nonce in ('short', 'x' * 129, 'bad nonce ' * 4):
|
|
response = _enroll(client, keyheaders, nonce=nonce)
|
|
assert response.status_code == 400, nonce
|
|
|
|
|
|
def test_enroll_requires_the_shared_key_while_it_is_allowed(client, db, app,
|
|
keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
response = client.post(ENROLL_URL, json=_enroll_body())
|
|
assert response.status_code in (401, 500)
|
|
|
|
|
|
def test_a_second_nonce_on_a_pending_request_stays_pending(client, db, app,
|
|
keyheaders):
|
|
from plugins.endpoints.models import EndpointEnrollRequest
|
|
|
|
_computer(db, 'BAY-01')
|
|
_setting(db, 'plugin.endpoints.autoenroll', False)
|
|
first = _enroll(client, keyheaders)
|
|
assert first.status_code == 202
|
|
|
|
# Auto-enroll back on: a different installation still does not get through.
|
|
_setting(db, 'plugin.endpoints.autoenroll', True)
|
|
second = _enroll(client, keyheaders, nonce=OTHER_NONCE)
|
|
assert second.status_code == 202
|
|
assert second.get_json()['data']['reason'] == 'nonce changed'
|
|
row = db.session.get(EndpointEnrollRequest,
|
|
first.get_json()['data']['requestid'])
|
|
assert row.enrollnonce == NONCE
|
|
|
|
|
|
def test_approved_request_issues_only_to_its_nonce(client, db, app, keyheaders,
|
|
auth_headers):
|
|
from plugins.endpoints.models import EndpointCredential
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
_setting(db, 'plugin.endpoints.autoenroll', False)
|
|
pending = _enroll(client, keyheaders)
|
|
requestid = pending.get_json()['data']['requestid']
|
|
|
|
approve = client.post(
|
|
f'/api/endpoints/enrollment/requests/{requestid}/approve',
|
|
json={'assetid': assetid}, headers=auth_headers)
|
|
assert approve.status_code == 200, approve.get_data(as_text=True)
|
|
|
|
wrong = _enroll(client, keyheaders, nonce=OTHER_NONCE)
|
|
assert wrong.status_code == 403
|
|
assert _code(wrong) == 'ENROLL_NONCE_MISMATCH'
|
|
|
|
right = _enroll(client, keyheaders)
|
|
assert right.status_code == 201, right.get_data(as_text=True)
|
|
assert EndpointCredential.query.filter_by(assetid=assetid).one() \
|
|
.enrolledvia == 'approved'
|
|
|
|
|
|
def test_shared_key_off_enroll_takes_no_auth_but_never_auto_enrolls(
|
|
client, db, app, keyheaders, auth_headers):
|
|
assetid = _computer(db, 'BAY-01')
|
|
_setting(db, 'plugin.endpoints.sharedkeyallowed', False)
|
|
|
|
pending = client.post(ENROLL_URL, json=_enroll_body())
|
|
assert pending.status_code == 202, pending.get_data(as_text=True)
|
|
assert pending.get_json()['data']['reason'] == 'awaiting approval'
|
|
|
|
requestid = pending.get_json()['data']['requestid']
|
|
client.post(f'/api/endpoints/enrollment/requests/{requestid}/approve',
|
|
json={'assetid': assetid}, headers=auth_headers)
|
|
issued = client.post(ENROLL_URL, json=_enroll_body())
|
|
assert issued.status_code == 201
|
|
|
|
|
|
def test_rejected_request_stays_rejected_for_the_same_nonce(
|
|
client, db, app, keyheaders, auth_headers):
|
|
_computer(db, 'BAY-01', serialnumber='SER-9')
|
|
pending = _enroll(client, keyheaders)
|
|
requestid = pending.get_json()['data']['requestid']
|
|
client.post(f'/api/endpoints/enrollment/requests/{requestid}/reject',
|
|
headers=auth_headers)
|
|
|
|
again = _enroll(client, keyheaders)
|
|
assert again.status_code == 202
|
|
assert again.get_json()['data']['reason'] == 'rejected'
|
|
|
|
|
|
# -- agent auth --------------------------------------------------------------
|
|
|
|
def test_shared_key_is_refused_after_enrollment(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
|
|
refused = client.get('/api/endpoints/alerts?hostname=BAY-01',
|
|
headers=keyheaders)
|
|
assert refused.status_code == 401
|
|
assert _code(refused) == 'SHARED_KEY_REFUSED'
|
|
|
|
ok = client.get('/api/endpoints/alerts',
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert ok.status_code == 200, ok.get_data(as_text=True)
|
|
|
|
|
|
def test_shared_key_still_works_for_a_pc_that_has_not_enrolled(
|
|
client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
_computer(db, 'BAY-02')
|
|
_credential(client, keyheaders)
|
|
|
|
response = client.get('/api/endpoints/alerts?hostname=BAY-02',
|
|
headers=keyheaders)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_shared_key_switched_off_is_refused(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
_setting(db, 'plugin.endpoints.sharedkeyallowed', False)
|
|
|
|
response = client.get('/api/endpoints/alerts?hostname=BAY-01',
|
|
headers=keyheaders)
|
|
assert response.status_code == 401
|
|
assert _code(response) == 'SHARED_KEY_REFUSED'
|
|
|
|
|
|
def test_credential_is_invalid_after_revoke(client, db, app, keyheaders,
|
|
auth_headers):
|
|
assetid = _computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
|
|
revoke = client.post(f'/api/endpoints/enrollment/assets/{assetid}/revoke',
|
|
headers=auth_headers)
|
|
assert revoke.status_code == 200, revoke.get_data(as_text=True)
|
|
|
|
response = client.get('/api/endpoints/alerts',
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 401
|
|
assert _code(response) == 'CREDENTIAL_INVALID'
|
|
|
|
# Revoked: the shared key is admitted again and the PC can re-enroll.
|
|
assert client.get('/api/endpoints/alerts?hostname=BAY-01',
|
|
headers=keyheaders).status_code == 200
|
|
reissued = _credential(client, 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'})
|
|
assert response.status_code == 401
|
|
assert _code(response) == 'CREDENTIAL_INVALID'
|
|
|
|
|
|
def test_package_downloads_need_no_identity(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
response = client.get('/api/endpoints/scriptbody/999',
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 404
|
|
|
|
|
|
# -- ASSET_MISMATCH ----------------------------------------------------------
|
|
|
|
def test_commands_next_refuses_another_hostname(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
_computer(db, 'BAY-02', serialnumber='SER-2')
|
|
secret = _credential(client, keyheaders)
|
|
|
|
response = client.get('/api/endpoints/commands/next?hostname=BAY-02',
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 403
|
|
assert _code(response) == 'ASSET_MISMATCH'
|
|
|
|
|
|
def test_commands_next_claims_for_the_credential_asset(client, db, app,
|
|
keyheaders):
|
|
from plugins.endpoints.models import EndpointCommand
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
db.session.add(EndpointCommand(assetid=assetid, action='mtconnect_status',
|
|
status=EndpointCommand.STATUS_PENDING))
|
|
db.session.commit()
|
|
|
|
response = client.get('/api/endpoints/commands/next',
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 200, response.get_data(as_text=True)
|
|
assert response.get_json()['data']['action'] == 'mtconnect_status'
|
|
|
|
|
|
def test_command_result_refuses_another_assets_command(client, db, app,
|
|
keyheaders):
|
|
from plugins.endpoints.models import EndpointCommand
|
|
|
|
_computer(db, 'BAY-01')
|
|
other = _computer(db, 'BAY-02', serialnumber='SER-2')
|
|
secret = _credential(client, keyheaders)
|
|
command = EndpointCommand(assetid=other, action='mtconnect_status',
|
|
status=EndpointCommand.STATUS_RUNNING)
|
|
db.session.add(command)
|
|
db.session.commit()
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/commands/{command.endpointcommandid}/result',
|
|
json={'succeeded': True},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 403
|
|
assert _code(response) == 'ASSET_MISMATCH'
|
|
|
|
|
|
def test_command_result_for_own_command_needs_no_hostname(client, db, app,
|
|
keyheaders):
|
|
from plugins.endpoints.models import EndpointCommand
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
command = EndpointCommand(assetid=assetid, action='mtconnect_status',
|
|
status=EndpointCommand.STATUS_RUNNING)
|
|
db.session.add(command)
|
|
db.session.commit()
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/commands/{command.endpointcommandid}/result',
|
|
json={'succeeded': True},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 200, response.get_data(as_text=True)
|
|
|
|
|
|
def _fleet_alert(db, alertscope=None):
|
|
from plugins.notifications.models import Notification, NotificationType
|
|
|
|
ntype = NotificationType(typename='Fleet Alert', typecolor='#d93f3f',
|
|
isactive=True)
|
|
db.session.add(ntype)
|
|
db.session.flush()
|
|
note = Notification(notificationtypeid=ntype.notificationtypeid,
|
|
notification='Line 3 is down', alertfleet=True,
|
|
alerttimeout=60, isactive=True, alertscope=alertscope)
|
|
db.session.add(note)
|
|
db.session.commit()
|
|
return note.notificationid
|
|
|
|
|
|
def test_alert_ack_refuses_another_asset(client, db, app, keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
other = _computer(db, 'BAY-02', serialnumber='SER-2')
|
|
secret = _credential(client, keyheaders)
|
|
notificationid = _fleet_alert(db)
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/alerts/{notificationid}/acknowledge',
|
|
json={'assetid': other},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 403
|
|
assert _code(response) == 'ASSET_MISMATCH'
|
|
|
|
byhostname = client.post(
|
|
f'/api/endpoints/alerts/{notificationid}/acknowledge',
|
|
json={'hostname': 'BAY-02'},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert byhostname.status_code == 403
|
|
assert _code(byhostname) == 'ASSET_MISMATCH'
|
|
|
|
|
|
def test_alert_ack_with_no_assetid_means_the_caller(client, db, app,
|
|
keyheaders):
|
|
from plugins.notifications.models import NotificationAcknowledgement
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
notificationid = _fleet_alert(db)
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/alerts/{notificationid}/acknowledge',
|
|
json={'assetid': 0},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 201, response.get_data(as_text=True)
|
|
assert NotificationAcknowledgement.query.filter_by(
|
|
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')
|
|
_computer(db, 'BAY-02', serialnumber='SER-2')
|
|
secret = _credential(client, keyheaders)
|
|
headers = {'X-Endpoint-Credential': secret}
|
|
|
|
mismatch = client.post('/api/collector/computers',
|
|
json={'hostname': 'BAY-02'}, headers=headers)
|
|
assert mismatch.status_code == 403, mismatch.get_data(as_text=True)
|
|
assert _code(mismatch) == 'ASSET_MISMATCH'
|
|
|
|
# A credential never creates a PC.
|
|
unknown = client.post('/api/collector/computers',
|
|
json={'hostname': 'NEW-BAY'}, headers=headers)
|
|
assert unknown.status_code == 403
|
|
assert _code(unknown) == 'ASSET_MISMATCH'
|
|
|
|
own = client.post('/api/collector/computers',
|
|
json={'hostname': 'BAY-01'}, headers=headers)
|
|
assert own.status_code == 200, own.get_data(as_text=True)
|
|
|
|
|
|
def test_collector_refuses_shared_key_for_an_enrolled_pc(client, db, app,
|
|
keyheaders):
|
|
_computer(db, 'BAY-01')
|
|
_credential(client, keyheaders)
|
|
|
|
response = client.post('/api/collector/computers',
|
|
json={'hostname': 'BAY-01'}, headers=keyheaders)
|
|
assert response.status_code == 401
|
|
assert _code(response) == 'SHARED_KEY_REFUSED'
|
|
|
|
|
|
def test_collector_invalid_credential(client, db, app):
|
|
response = client.post('/api/collector/printers',
|
|
json={'hostname': 'BAY-01', 'queues': []},
|
|
headers={'X-Endpoint-Credential': 'sdbe_' + 'a' * 43})
|
|
assert response.status_code == 401
|
|
assert _code(response) == 'CREDENTIAL_INVALID'
|
|
|
|
|
|
def test_endpoints_collect_route_is_the_collector(client, db, app, keyheaders):
|
|
from plugins.endpoints.models import EndpointState
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
secret = _credential(client, keyheaders)
|
|
|
|
response = client.post('/api/endpoints/collect',
|
|
json={'hostname': 'BAY-01', 'agentversion': '1.20.0'},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert response.status_code == 200, response.get_data(as_text=True)
|
|
assert EndpointState.query.filter_by(assetid=assetid).one() \
|
|
.agentversion == '1.20.0'
|
|
|
|
_computer(db, 'BAY-02', serialnumber='SER-2')
|
|
mismatch = client.post('/api/endpoints/collect',
|
|
json={'hostname': 'BAY-02'},
|
|
headers={'X-Endpoint-Credential': secret})
|
|
assert mismatch.status_code == 403
|
|
assert _code(mismatch) == 'ASSET_MISMATCH'
|
|
|
|
|
|
def test_contract_exports_answer_through_the_endpoints_plugin(client, db, app,
|
|
keyheaders):
|
|
from shopdb.api import agent_shared_key_refused, resolve_agent_credential
|
|
|
|
assetid = _computer(db, 'BAY-01')
|
|
assert agent_shared_key_refused('BAY-01') is False
|
|
secret = _credential(client, keyheaders)
|
|
with app.test_request_context():
|
|
assert resolve_agent_credential(secret) == assetid
|
|
assert resolve_agent_credential('sdbe_' + 'z' * 43) is None
|
|
assert agent_shared_key_refused('BAY-01') is True
|
|
assert agent_shared_key_refused('UNKNOWN') is False
|
|
|
|
|
|
# -- admin gating ------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize('method,url', [
|
|
('get', '/api/endpoints/enrollment'),
|
|
('get', '/api/endpoints/enrollment/migration'),
|
|
('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'),
|
|
])
|
|
def test_member_without_a_role_is_forbidden(client, db, member_headers, method,
|
|
url):
|
|
response = getattr(client, method)(url, json={}, headers=member_headers)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_enrollment_listing_and_settings(client, db, app, keyheaders,
|
|
auth_headers):
|
|
enrolled = _computer(db, 'BAY-01')
|
|
_computer(db, 'BAY-02', serialnumber='SER-2')
|
|
_computer(db, 'BAY-03', serialnumber=None)
|
|
_credential(client, keyheaders)
|
|
_enroll(client, keyheaders, hostname='BAY-03', serialnumber='X')
|
|
|
|
response = client.get('/api/endpoints/enrollment', headers=auth_headers)
|
|
assert response.status_code == 200, response.get_data(as_text=True)
|
|
body = response.get_json()
|
|
data = body['data']
|
|
assert data['settings'] == {'sharedkeyallowed': True, 'autoenroll': True}
|
|
assert data['counts'] == {'enrolled': 1, 'notenrolled': 1, 'pending': 1}
|
|
states = {row['hostname']: row['state'] for row in data['computers']}
|
|
assert states == {'BAY-01': 'enrolled', 'BAY-02': 'notenrolled',
|
|
'BAY-03': 'pending'}
|
|
row = next(r for r in data['computers'] if r['assetid'] == enrolled)
|
|
assert row['credentialprefix'].startswith('sdbe_')
|
|
assert row['enrolledvia'] == 'auto'
|
|
assert [r['reason'] for r in data['requests']
|
|
if r['status'] == 'pending'] == ['no serial on record']
|
|
assert body['meta']['pagination']['total'] == 3
|
|
|
|
filtered = client.get('/api/endpoints/enrollment?state=enrolled',
|
|
headers=auth_headers).get_json()['data']
|
|
assert [r['hostname'] for r in filtered['computers']] == ['BAY-01']
|
|
|
|
put = client.put('/api/endpoints/enrollment/settings',
|
|
json={'sharedkeyallowed': False}, headers=auth_headers)
|
|
assert put.status_code == 200
|
|
assert put.get_json()['data'] == {'sharedkeyallowed': False,
|
|
'autoenroll': True}
|
|
bad = client.put('/api/endpoints/enrollment/settings',
|
|
json={'autoenroll': 'yes'}, headers=auth_headers)
|
|
assert bad.status_code == 400
|
|
|
|
|
|
def test_approve_refuses_an_enrolled_computer(client, db, app, keyheaders,
|
|
auth_headers):
|
|
assetid = _computer(db, 'BAY-01')
|
|
_credential(client, keyheaders)
|
|
pending = _enroll(client, keyheaders, nonce=OTHER_NONCE)
|
|
requestid = pending.get_json()['data']['requestid']
|
|
|
|
response = client.post(
|
|
f'/api/endpoints/enrollment/requests/{requestid}/approve',
|
|
json={'assetid': assetid}, headers=auth_headers)
|
|
assert response.status_code == 409
|
|
|
|
|
|
# -- server certificate pin --------------------------------------------------
|
|
|
|
def test_server_cert_reports_the_spki_pin(client, db, auth_headers, tmp_path,
|
|
monkeypatch):
|
|
import hashlib
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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 routes as endpoint_routes
|
|
|
|
key = ec.generate_private_key(ec.SECP256R1())
|
|
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'pin.example.test')])
|
|
now = datetime.now(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 - timedelta(days=1))
|
|
.not_valid_after(now + timedelta(days=30))
|
|
.sign(key, hashes.SHA256()))
|
|
certdir = tmp_path / 'certs'
|
|
certdir.mkdir()
|
|
(certdir / 'server.crt').write_bytes(
|
|
certificate.public_bytes(serialization.Encoding.PEM))
|
|
monkeypatch.chdir(tmp_path)
|
|
if hasattr(endpoint_routes, '_server_cert_dir'):
|
|
monkeypatch.setattr(endpoint_routes, '_server_cert_dir', lambda: certdir)
|
|
|
|
expected = hashlib.sha256(key.public_key().public_bytes(
|
|
serialization.Encoding.DER,
|
|
serialization.PublicFormat.SubjectPublicKeyInfo)).hexdigest()
|
|
|
|
response = client.get('/api/endpoints/settings/server-cert',
|
|
headers=auth_headers)
|
|
assert response.status_code == 200, response.get_data(as_text=True)
|
|
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
|