Show per-PC credentials across the admin interface
- Endpoints (shared key) page: the key is presented as what PCs use until they enroll, with enrollment progress, a link to Enrollment, rewritten guidance, and rotate/revoke confirmations that say how many PCs are not enrolled yet. - Settings cards describe the server certificate as what PCs actually receive, on IIS or Docker; Enrollment has its own icon (the icon set lacked the one the Server Certificate card already asked for). - Server Certificate page checks the certificate the site presents at its public address and compares its pin with the certificate file, warning when they differ. The host comes from site settings only; the check is a TLS handshake, never an HTTP request. - Fleet list: Credential column and filter (enrolled, pending, re-enroll blocked, shared key, not reporting), computed in the list query. - PC detail: credential state, prefix and dates for fleet readers, state only for others, with admin Revoke (optionally blocking re-enrollment) and Allow re-enrollment. - Dashboard: pending enrollment requests, and PCs blocked from re-enrolling for more than seven days. - Runbook: where to find each of these in the app.
This commit is contained in:
cproudlock
@@ -73,6 +73,8 @@ def test_a_viewall_target_is_one_of_the_known_pages(app):
|
||||
'/pcs', '/machines', '/printers', '/network', '/notifications',
|
||||
'/geenforce', '/reports/toner', '/reports/warranty',
|
||||
'/reports/pc-relationships', '/reports/calibration', '/reports',
|
||||
# plugins/endpoints/frontend/routes.js, readable by endpoints.fleet
|
||||
'/settings/endpoints/enrollment',
|
||||
}
|
||||
with app.app_context():
|
||||
for pluginname, card in _declared(app):
|
||||
|
||||
525
tests/test_plugins/test_endpoints_credential_admin_ui.py
Normal file
525
tests/test_plugins/test_endpoints_credential_admin_ui.py
Normal file
@@ -0,0 +1,525 @@
|
||||
"""Admin surfaces for per-PC endpoint credentials.
|
||||
|
||||
The fleet list, the PC detail status and the dashboard all describe the same
|
||||
fact - which credential a PC reports with - so they share one state vocabulary:
|
||||
enrolled, reenrollblocked, pending, notreporting, sharedkey. The list computes
|
||||
it in SQL so a filter survives pagination; PC detail redacts everything but the
|
||||
state for a reader below endpoints.fleet; the server certificate live check
|
||||
reads the host from settings, never from the request.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Setting
|
||||
|
||||
|
||||
def _now():
|
||||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _pc(db, hostname, reported=True, serialnumber='SER'):
|
||||
from plugins.computers.models import Computer
|
||||
from plugins.endpoints.models import EndpointState
|
||||
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if atype is None:
|
||||
atype = AssetType(assettype='computer', pluginname='computers',
|
||||
tablename='computers')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
asset = Asset(assetnumber=hostname, name=hostname, serialnumber=serialnumber,
|
||||
assettypeid=atype.assettypeid, isactive=True)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
computer = Computer(assetid=asset.assetid, hostname=hostname)
|
||||
db.session.add(computer)
|
||||
db.session.add(EndpointState(
|
||||
assetid=asset.assetid, hostname=hostname, agentversion='1.20.0',
|
||||
lastreportedat=_now() if reported else None))
|
||||
db.session.commit()
|
||||
return asset.assetid
|
||||
|
||||
|
||||
def _credential(db, assetid, revoked=False, blocked=False, reason=None,
|
||||
revokedat=None, prefix='sdbe_abcdefg'):
|
||||
from plugins.endpoints.models import EndpointCredential
|
||||
|
||||
row = EndpointCredential(
|
||||
assetid=assetid, credentialhash=hashlib.sha256(
|
||||
str(assetid).encode()).hexdigest(),
|
||||
credentialprefix=prefix, enrolledvia='auto', issuedat=_now(),
|
||||
lastusedat=_now(),
|
||||
revokedat=(revokedat or _now()) if revoked else None,
|
||||
revokedby='admin' if revoked else None,
|
||||
reenrollblocked=blocked, reenrollblockedreason=reason)
|
||||
db.session.add(row)
|
||||
db.session.commit()
|
||||
return row
|
||||
|
||||
|
||||
def _request(db, hostname, matchedassetid=None, reason='awaiting approval',
|
||||
status='pending'):
|
||||
from plugins.endpoints.models import EndpointEnrollRequest
|
||||
|
||||
row = EndpointEnrollRequest(
|
||||
hostname=hostname, serialnumber='SER', enrollnonce='n' * 40,
|
||||
status=status, reason=reason, matchedassetid=matchedassetid,
|
||||
requestedat=_now(), lastseenat=_now())
|
||||
db.session.add(row)
|
||||
db.session.commit()
|
||||
return row
|
||||
|
||||
|
||||
def _login(db, client, username, permissions=()):
|
||||
from shopdb.core.models import User, Role, Permission
|
||||
|
||||
Permission.seed()
|
||||
db.session.flush()
|
||||
role = Role(rolename=username + 'role')
|
||||
db.session.add(role)
|
||||
db.session.flush()
|
||||
for name in permissions:
|
||||
perm = Permission.query.filter_by(name=name).first()
|
||||
assert perm is not None, name
|
||||
role.permissions.append(perm)
|
||||
user = User(username=username, email=username + '@test.local',
|
||||
passwordhash=generate_password_hash('testpass'))
|
||||
user.roles.append(role)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
response = client.post('/api/auth/login',
|
||||
json={'username': username, 'password': 'testpass'})
|
||||
return {'Authorization': 'Bearer ' + response.get_json()['data']['access_token']}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fleet(db):
|
||||
"""One PC in every credential state."""
|
||||
ids = {
|
||||
'ENROLLED': _pc(db, 'ENROLLED'),
|
||||
'BLOCKED': _pc(db, 'BLOCKED'),
|
||||
'PENDING': _pc(db, 'PENDING'),
|
||||
'SILENT': _pc(db, 'SILENT', reported=False),
|
||||
'SHARED': _pc(db, 'SHARED'),
|
||||
'REVOKED': _pc(db, 'REVOKED'),
|
||||
}
|
||||
_credential(db, ids['ENROLLED'])
|
||||
_credential(db, ids['BLOCKED'], revoked=True, blocked=True, reason='rollback')
|
||||
# A blocked PC keeps asking; its request must not turn it into pending.
|
||||
_request(db, 'BLOCKED', ids['BLOCKED'], reason='reenroll blocked')
|
||||
_request(db, 'pending')
|
||||
_credential(db, ids['REVOKED'], revoked=True)
|
||||
return ids
|
||||
|
||||
|
||||
EXPECTED = {
|
||||
'ENROLLED': 'enrolled',
|
||||
'BLOCKED': 'reenrollblocked',
|
||||
'PENDING': 'pending',
|
||||
'SILENT': 'notreporting',
|
||||
'SHARED': 'sharedkey',
|
||||
'REVOKED': 'sharedkey',
|
||||
}
|
||||
|
||||
|
||||
# -- fleet list ---------------------------------------------------------------
|
||||
|
||||
def test_fleet_list_carries_the_credential_state(client, db, auth_headers, fleet):
|
||||
response = client.get('/api/endpoints?perpage=100', headers=auth_headers)
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
states = {row['hostname']: row['credentialstate']
|
||||
for row in response.get_json()['data']}
|
||||
assert states == EXPECTED
|
||||
|
||||
|
||||
@pytest.mark.parametrize('state', sorted(set(EXPECTED.values())))
|
||||
def test_fleet_list_filters_by_credential_state(client, db, auth_headers, fleet,
|
||||
state):
|
||||
response = client.get(f'/api/endpoints?credentialstate={state}',
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
body = response.get_json()
|
||||
hosts = sorted(row['hostname'] for row in body['data'])
|
||||
assert hosts == sorted(h for h, s in EXPECTED.items() if s == state)
|
||||
assert body['meta']['pagination']['total'] == len(hosts)
|
||||
|
||||
|
||||
def test_credential_filter_keeps_paging(client, db, auth_headers, fleet):
|
||||
response = client.get('/api/endpoints?credentialstate=sharedkey&perpage=1'
|
||||
'&sort=hostname', headers=auth_headers)
|
||||
body = response.get_json()
|
||||
assert [row['hostname'] for row in body['data']] == ['REVOKED']
|
||||
assert body['meta']['pagination']['total'] == 2
|
||||
|
||||
|
||||
def test_fleet_list_state_costs_no_query_per_row(client, db, app, auth_headers,
|
||||
fleet):
|
||||
from sqlalchemy import event
|
||||
|
||||
statements = []
|
||||
|
||||
def record(_connection, _cursor, statement, *_args):
|
||||
if 'endpointcredentials' in statement or 'endpointenrollrequests' in statement:
|
||||
statements.append(statement)
|
||||
|
||||
engine = db.engine
|
||||
event.listen(engine, 'before_cursor_execute', record)
|
||||
try:
|
||||
client.get('/api/endpoints?perpage=100', headers=auth_headers)
|
||||
small = len(statements)
|
||||
for index in range(10):
|
||||
_pc(db, f'EXTRA-{index:02d}')
|
||||
statements.clear()
|
||||
client.get('/api/endpoints?perpage=100', headers=auth_headers)
|
||||
large = len(statements)
|
||||
finally:
|
||||
event.remove(engine, 'before_cursor_execute', record)
|
||||
assert small == large
|
||||
assert large <= 2
|
||||
|
||||
|
||||
def test_unknown_credential_filter_is_ignored_like_status(client, db,
|
||||
auth_headers, fleet):
|
||||
response = client.get('/api/endpoints?credentialstate=bogus',
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert len(response.get_json()['data']) == len(EXPECTED)
|
||||
|
||||
|
||||
# -- PC detail ----------------------------------------------------------------
|
||||
|
||||
def test_endpoint_detail_credential_for_fleet_holders(client, db, auth_headers,
|
||||
fleet):
|
||||
data = client.get(f"/api/endpoints/{fleet['BLOCKED']}",
|
||||
headers=auth_headers).get_json()['data']
|
||||
credential = data['credential']
|
||||
assert credential['state'] == 'reenrollblocked'
|
||||
assert credential['statelabel'] == 'Re-enroll blocked'
|
||||
assert credential['credentialprefix'] == 'sdbe_abcdefg'
|
||||
assert credential['issuedat'] and credential['lastusedat']
|
||||
assert credential['enrolledvia'] == 'auto'
|
||||
assert credential['reenrollblocked'] is True
|
||||
assert credential['reenrollblockedreason'] == 'rollback'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('hostname', sorted(EXPECTED))
|
||||
def test_endpoint_detail_state_matches_the_list(client, db, auth_headers, fleet,
|
||||
hostname):
|
||||
data = client.get(f'/api/endpoints/{fleet[hostname]}',
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert data['credential']['state'] == EXPECTED[hostname]
|
||||
|
||||
|
||||
def test_endpoint_detail_redacts_below_fleet(client, db, fleet):
|
||||
headers = _login(db, client, 'viewer', ['endpoints.view'])
|
||||
response = client.get(f"/api/endpoints/{fleet['ENROLLED']}", headers=headers)
|
||||
assert response.status_code == 200
|
||||
credential = response.get_json()['data']['credential']
|
||||
assert credential['state'] == 'enrolled'
|
||||
for key in ('credentialprefix', 'issuedat', 'lastusedat', 'revokedat',
|
||||
'enrolledvia', 'reenrollblockedreason'):
|
||||
assert credential[key] is None, key
|
||||
|
||||
panel = client.get(f"/api/endpoints/asset/{fleet['ENROLLED']}/status",
|
||||
headers=headers).get_json()['data']
|
||||
labels = {field['label']: field['value'] for field in panel['fields']}
|
||||
assert labels['Credential'] == 'Enrolled'
|
||||
assert 'Credential prefix' not in labels
|
||||
assert 'Credential issued' not in labels
|
||||
assert 'Credential last used' not in labels
|
||||
|
||||
|
||||
def test_guest_with_endpoints_view_sees_state_only(client, db, fleet):
|
||||
from shopdb.core.models import Permission, Role
|
||||
|
||||
Permission.seed()
|
||||
Role.seedguest()
|
||||
db.session.commit()
|
||||
guest = Role.guest()
|
||||
guest.permissions = Permission.query.filter_by(name='endpoints.view').all()
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(f"/api/endpoints/asset/{fleet['BLOCKED']}/status")
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
data = response.get_json()['data']
|
||||
labels = {field['label']: field['value'] for field in data['fields']}
|
||||
assert labels['Credential'] == 'Re-enroll blocked'
|
||||
assert 'Re-enroll block reason' not in labels
|
||||
assert data['credential']['credentialprefix'] is None
|
||||
assert data['credential']['reenrollblockedreason'] is None
|
||||
|
||||
|
||||
def test_status_panel_shows_credential_fields_to_fleet_holders(client, db,
|
||||
auth_headers,
|
||||
fleet):
|
||||
panel = client.get(f"/api/endpoints/asset/{fleet['BLOCKED']}/status",
|
||||
headers=auth_headers).get_json()['data']
|
||||
labels = {field['label']: field for field in panel['fields']}
|
||||
assert labels['Credential']['value'] == 'Re-enroll blocked'
|
||||
assert labels['Credential prefix']['value'] == 'sdbe_abcdefg...'
|
||||
assert labels['Credential prefix']['mono'] is True
|
||||
assert labels['Credential issued']['format'] == 'datetime'
|
||||
assert labels['Credential last used']['format'] == 'datetime'
|
||||
assert labels['Enrolled via']['value'] == 'auto'
|
||||
assert labels['Re-enroll block reason']['value'] == 'rollback'
|
||||
|
||||
|
||||
# -- server certificate live check -------------------------------------------
|
||||
|
||||
def _certificate_pem_and_der():
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.x509.oid import NameOID
|
||||
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'live.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()))
|
||||
spki = key.public_key().public_bytes(
|
||||
serialization.Encoding.DER,
|
||||
serialization.PublicFormat.SubjectPublicKeyInfo)
|
||||
return (certificate.public_bytes(serialization.Encoding.PEM),
|
||||
certificate.public_bytes(serialization.Encoding.DER),
|
||||
hashlib.sha256(spki).hexdigest(),
|
||||
certificate.fingerprint(hashes.SHA1()).hex().upper())
|
||||
|
||||
|
||||
class _FakeSocket:
|
||||
def __init__(self, der=None):
|
||||
self.der = der
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args):
|
||||
return False
|
||||
|
||||
def getpeercert(self, binary_form=False):
|
||||
assert binary_form is True
|
||||
return self.der
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tls(monkeypatch, tmp_path):
|
||||
"""Replace the socket and ssl layer; record what the route connected to."""
|
||||
import socket
|
||||
import ssl
|
||||
|
||||
from plugins.endpoints.api import routes as endpoint_routes
|
||||
|
||||
certdir = tmp_path / 'certs'
|
||||
certdir.mkdir()
|
||||
monkeypatch.setattr(endpoint_routes, '_server_cert_dir', lambda: certdir)
|
||||
|
||||
state = {'der': None, 'connected': [], 'wrapped': [], 'error': None}
|
||||
|
||||
def create_connection(address, timeout=None):
|
||||
state['connected'].append((address, timeout))
|
||||
if state['error'] is not None:
|
||||
raise state['error']
|
||||
return _FakeSocket()
|
||||
|
||||
def wrap_socket(self, sock, server_hostname=None, **_kwargs):
|
||||
state['wrapped'].append((server_hostname, self.verify_mode,
|
||||
self.check_hostname))
|
||||
return _FakeSocket(state['der'])
|
||||
|
||||
monkeypatch.setattr(socket, 'create_connection', create_connection)
|
||||
monkeypatch.setattr(ssl.SSLContext, 'wrap_socket', wrap_socket)
|
||||
state['certdir'] = certdir
|
||||
return state
|
||||
|
||||
|
||||
LIVE_URL = '/api/endpoints/settings/server-cert/live'
|
||||
|
||||
|
||||
def test_live_check_unconfigured(client, db, auth_headers, tls):
|
||||
response = client.get(LIVE_URL, headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()['data'] == {'configured': False}
|
||||
assert tls['connected'] == []
|
||||
|
||||
|
||||
def test_live_check_matches_the_file(client, db, auth_headers, tls):
|
||||
import ssl
|
||||
|
||||
pem, der, pin, sha1 = _certificate_pem_and_der()
|
||||
(tls['certdir'] / 'server.crt').write_bytes(pem)
|
||||
tls['der'] = der
|
||||
Setting.set('site_base_url', 'https://shopdb.example.test:8443/shopdb')
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(LIVE_URL, headers={**auth_headers,
|
||||
'Host': 'attacker.example'})
|
||||
assert response.status_code == 200, response.get_data(as_text=True)
|
||||
data = response.get_json()['data']
|
||||
assert data == {
|
||||
'configured': True,
|
||||
'url': 'https://shopdb.example.test:8443/shopdb',
|
||||
'presentedspkisha256': pin,
|
||||
'presentedsha1thumbprint': sha1,
|
||||
'matchesfile': True,
|
||||
'error': None,
|
||||
}
|
||||
# Host from the setting, never the request; 5 s; read-only handshake.
|
||||
assert tls['connected'] == [(('shopdb.example.test', 8443), 5)]
|
||||
assert tls['wrapped'] == [('shopdb.example.test', ssl.CERT_NONE, False)]
|
||||
|
||||
|
||||
def test_live_check_reports_a_different_certificate(client, db, auth_headers,
|
||||
tls):
|
||||
filepem, _der, filepin, _sha1 = _certificate_pem_and_der()
|
||||
_pem, liveder, livepin, _livesha1 = _certificate_pem_and_der()
|
||||
(tls['certdir'] / 'server.crt').write_bytes(filepem)
|
||||
tls['der'] = liveder
|
||||
Setting.set('site_base_url', 'https://shopdb.example.test')
|
||||
db.session.commit()
|
||||
|
||||
data = client.get(LIVE_URL, headers=auth_headers).get_json()['data']
|
||||
assert data['presentedspkisha256'] == livepin != filepin
|
||||
assert data['matchesfile'] is False
|
||||
assert data['error'] is None
|
||||
assert tls['connected'][0][0] == ('shopdb.example.test', 443)
|
||||
|
||||
|
||||
def test_live_check_no_file_never_matches(client, db, auth_headers, tls):
|
||||
_pem, der, pin, _sha1 = _certificate_pem_and_der()
|
||||
tls['der'] = der
|
||||
Setting.set('site_base_url', 'https://10.0.0.5')
|
||||
db.session.commit()
|
||||
|
||||
data = client.get(LIVE_URL, headers=auth_headers).get_json()['data']
|
||||
assert data['presentedspkisha256'] == pin
|
||||
assert data['matchesfile'] is False
|
||||
# No SNI for an address.
|
||||
assert tls['wrapped'][0][0] is None
|
||||
|
||||
|
||||
def test_live_check_connection_failure(client, db, auth_headers, tls):
|
||||
tls['error'] = OSError('Connection refused')
|
||||
Setting.set('site_base_url', 'https://shopdb.example.test')
|
||||
db.session.commit()
|
||||
|
||||
response = client.get(LIVE_URL, headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()['data']
|
||||
assert data['configured'] is True
|
||||
assert data['presentedspkisha256'] is None
|
||||
assert data['matchesfile'] is False
|
||||
assert 'Connection refused' in data['error']
|
||||
|
||||
|
||||
def test_live_check_refuses_plain_http(client, db, auth_headers, tls):
|
||||
Setting.set('site_base_url', 'http://shopdb.example.test')
|
||||
db.session.commit()
|
||||
|
||||
data = client.get(LIVE_URL, headers=auth_headers).get_json()['data']
|
||||
assert data['configured'] is True
|
||||
assert 'https' in data['error']
|
||||
assert tls['connected'] == []
|
||||
|
||||
|
||||
def test_live_check_allows_fleet_holders(client, db, tls):
|
||||
headers = _login(db, client, 'fleetreader', ['endpoints.fleet'])
|
||||
assert client.get(LIVE_URL, headers=headers).status_code == 200
|
||||
|
||||
|
||||
# -- dashboard ----------------------------------------------------------------
|
||||
|
||||
def test_pending_enrollment_attention_row(client, db, auth_headers, fleet):
|
||||
response = client.get('/api/endpoints/dashboard/enrollmentpending',
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
# The blocked PC's own request is expected noise, not work.
|
||||
assert response.get_json()['data'] == [
|
||||
{'summary': '1 PC waiting for enrollment approval', 'pendingcount': 1}]
|
||||
|
||||
_request(db, 'ANOTHER')
|
||||
data = client.get('/api/endpoints/dashboard/enrollmentpending',
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert data[0]['summary'] == '2 PCs waiting for enrollment approval'
|
||||
|
||||
|
||||
def test_pending_enrollment_attention_hides_when_nothing_waits(client, db,
|
||||
auth_headers):
|
||||
response = client.get('/api/endpoints/dashboard/enrollmentpending',
|
||||
headers=auth_headers)
|
||||
assert response.get_json()['data'] == []
|
||||
|
||||
|
||||
def test_reenroll_blocked_attention_after_seven_days(client, db, auth_headers):
|
||||
from plugins.computers.models import Computer
|
||||
|
||||
fresh = _pc(db, 'FRESH')
|
||||
stale = _pc(db, 'STALE')
|
||||
_credential(db, fresh, revoked=True, blocked=True,
|
||||
revokedat=_now() - timedelta(days=2))
|
||||
_credential(db, stale, revoked=True, blocked=True, reason='rollback',
|
||||
revokedat=_now() - timedelta(days=9))
|
||||
|
||||
data = client.get('/api/endpoints/dashboard/reenrollblocked',
|
||||
headers=auth_headers).get_json()['data']
|
||||
computerid = Computer.query.filter_by(assetid=stale).one().computerid
|
||||
assert data == [{'computerid': computerid, 'assetid': stale,
|
||||
'hostname': 'STALE', 'blockeddays': 9,
|
||||
'reason': 'rollback'}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('url', [
|
||||
'/api/endpoints/dashboard/enrollmentpending',
|
||||
'/api/endpoints/dashboard/reenrollblocked',
|
||||
LIVE_URL,
|
||||
])
|
||||
def test_new_reads_need_a_permission(client, db, member_headers, url):
|
||||
assert client.get(url, headers=member_headers).status_code == 403
|
||||
assert client.get(url).status_code == 401
|
||||
|
||||
|
||||
def test_dashboard_widgets_declare_the_enrollment_items():
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
|
||||
widgets = {w['id']: w for w in EndpointsPlugin().get_dashboard_widgets()}
|
||||
pending = widgets['endpoints-enrollment-pending']
|
||||
assert pending['severity'] == 'warning'
|
||||
assert pending['endpoint'] == '/api/endpoints/dashboard/enrollmentpending'
|
||||
assert pending['map']['link'] == '/settings/endpoints/enrollment'
|
||||
blocked = widgets['endpoints-reenroll-blocked']
|
||||
assert blocked['severity'] == 'info'
|
||||
assert blocked['map']['link'] == '/pcs/{computerid}'
|
||||
for widget in (pending, blocked):
|
||||
assert widget['permission'] == 'endpoints.fleet'
|
||||
assert widget['empty'] == 'hide'
|
||||
|
||||
|
||||
# -- settings cards -----------------------------------------------------------
|
||||
|
||||
def test_settings_cards_describe_the_new_model_and_their_icons_render():
|
||||
from plugins.endpoints.plugin import EndpointsPlugin
|
||||
|
||||
cards = {card['to']: card for card in EndpointsPlugin().get_settings_cards()}
|
||||
cert = cards['/settings/endpoints/server-cert']
|
||||
assert 'nginx' not in cert['description'].lower()
|
||||
assert 'IIS' in cert['description'] and 'Docker' in cert['description']
|
||||
assert 'shared' in cards['/settings/endpoints/agentkey']['description']
|
||||
enrollment = cards['/settings/endpoints/enrollment']
|
||||
assert enrollment['icon'] != cards['/settings/endpoints/agentkey']['icon']
|
||||
|
||||
catalog = os.path.join(os.path.dirname(__file__), '..', '..', 'frontend',
|
||||
'src', 'composables', 'settingsCatalog.js')
|
||||
with open(os.path.normpath(catalog), encoding='utf-8') as handle:
|
||||
source = handle.read()
|
||||
for card in cards.values():
|
||||
icon = card['icon']
|
||||
assert f"'{icon}':" in source or f" {icon}:" in source, icon
|
||||
Reference in New Issue
Block a user