7480e4fe3a
Per-PC credentials (replaces the 09-14 client-certificate attempt): - POST /api/endpoints/enroll issues a credential to a PC that matches exactly one computer by hostname and BIOS serial and has none yet. Anything else (unknown or duplicate hostname, serial mismatch, already enrolled) waits as a request an admin approves or rejects. The agent's persisted nonce binds an approved request to the PC that asked. - One auth helper serves every agent route. With X-Endpoint-Credential the asset comes from the credential and a request naming another PC gets 403 ASSET_MISMATCH. The shared key still works for PCs that have not enrolled, is refused (401 SHARED_KEY_REFUSED) for any PC that has, and can be switched off. A revoked credential gets 401 CREDENTIAL_INVALID so the agent re-enrolls. - The core collector accepts the credential for the computers, printers and endpoints collectors through two new plugin hooks exposed on shopdb.api (contract 0.33.0), and refuses the shared key for enrolled PCs. - Admin API: enrollment list, approve, reject, revoke (un-enroll), settings. Migration API: per-PC readiness with blockers, and upgrade waves that package the agent once and queue the existing agent upgrade per PC. - Migration 0023 creates endpointcredentials, endpointenrollrequests and endpointmigrationwaves, drops the certificate columns from assets where test databases have them, and renames app_fingerprint to appfingerprint. 0021 and 0022 keep their ids but no longer alter the core assets table. - Removed: client_cert.py, certificate renewal, the unauthenticated /test/client-cert route, nginx client certificate forwarding and the agent certificate script. The client-cert header could be forged on IIS, the key was readable by users, and the shared key remained a fallback. - GET /settings/server-cert returns the SPKI SHA-256 pin for GE-Enforce. Review fixes: - report_command_result used g without importing it; every result was a 500. - collect_log refuses setup answer files, appsettings, agent directories and key files at enqueue. - Fleet alerts: alerttimeout and alertfleet agree with the migration, the scope is validated and a malformed scope no longer reaches the whole fleet, and the ack race no longer raises NameError. - A refused break-glass override leaves no hold and no audit row. - Agent upgrade checks locks and permissions before repackaging, and writes the package atomically. - Server certificate upload validates the pair before replacing files; generation no longer builds an openssl config from user input; audit rows are actually written. - Health evaluation runs in a savepoint so a caught error cannot lose the check-in. - Required-app policies are found by appid, survive renames, and deleting an app deactivates policies that still name it. The UI and API now say such a policy only takes effect inside a business policy. - Backfill handles duplicate fingerprints and rolls back a failed batch. - Plugin settings routes skip a plugin whose hook raises. - The setup token is no longer printed to the log. - The naming gate checks top-level files again and handles odd file names. - Removed root scratch scripts, one of which carried a token. - Rewrote the fleet alert scheduling and app delta tests on real fixtures.
590 lines
23 KiB
Python
590 lines
23 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_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):
|
|
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)
|
|
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_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'),
|
|
('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)
|
|
assert response.get_json()['data']['spkisha256'] == expected
|