GE-Enforce: compliance view, inline payload upload, frontend test harness
Fleet-install compliance for app-linked manifest entries: new service compliance_for_scope + GET /geenforce/scopes/<id>/compliance count active ComputerInstalledApp rows by curated appid (null-safe when computers plugin absent). ManifestEditor gains a compliance panel. Curated appid stays shopdb metadata and never enters manifest JSON, so behavioral parity is unaffected. Inline manifest payloads: store_inline_payload (sha256, 1MB cap, payloadsource='inline') + POST/GET /geenforce/entries/<id>/payload; editor gains an upload control. Entry payload metadata surfaced in _entry_payload. Frontend test harness: extract the editor's entry-form logic into pure entryForm.js (buildEntryPayload, describeEntry, availableEntryTypes, scope gates, ...) and cover it with 45 vitest tests. ManifestEditor now imports those helpers, so the tests exercise the shipped code path (no duplication). 908 backend tests pass; vitest 45 pass; frontend build green; naming green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
166
tests/test_plugins/test_geenforce_compliance.py
Normal file
166
tests/test_plugins/test_geenforce_compliance.py
Normal file
@@ -0,0 +1,166 @@
|
||||
"""GE-Enforce compliance: fleet-install coverage per app-linked entry.
|
||||
|
||||
Compliance pairs each app-linked manifest entry with collected PC data
|
||||
(ComputerInstalledApp) to answer 'how many PCs have this app, and how many on
|
||||
the expected version'. Only entries carrying a curated appid appear; unlinked
|
||||
entries are skipped. FileVersion detection supplies the expected-version target;
|
||||
every other detection method yields a null target (no meaningful match). When
|
||||
the optional computers plugin is absent, counts degrade to null gracefully.
|
||||
"""
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Application
|
||||
|
||||
from plugins.geenforce import service
|
||||
|
||||
|
||||
def _seed_app(db, appname):
|
||||
app_row = Application(appname=appname)
|
||||
db.session.add(app_row)
|
||||
db.session.commit()
|
||||
return app_row.appid
|
||||
|
||||
|
||||
def _seed_pc_with_app(db, hostname, appid, installedversion, isactive=True):
|
||||
"""Create a Computer + one ComputerInstalledApp row for it."""
|
||||
from plugins.computers.models import Computer, ComputerInstalledApp
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if not atype:
|
||||
atype = AssetType(assettype='computer')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
asset = Asset(assetnumber=f'PC-{hostname}', assettypeid=atype.assettypeid)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
comp = Computer(assetid=asset.assetid, hostname=hostname)
|
||||
db.session.add(comp)
|
||||
db.session.flush()
|
||||
link = ComputerInstalledApp(computerid=comp.computerid, appid=appid,
|
||||
installedversion=installedversion,
|
||||
isactive=isactive)
|
||||
db.session.add(link)
|
||||
db.session.commit()
|
||||
return comp
|
||||
|
||||
|
||||
def _create_scope(client, auth_headers, name='gea-shopfloor-cmm'):
|
||||
resp = client.post('/api/geenforce/scopes',
|
||||
json={'scopename': name, 'phase': 'runtime'},
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()['data']['scopeid']
|
||||
|
||||
|
||||
def _add_entry(client, auth_headers, scopeid, entry):
|
||||
resp = client.post(f'/api/geenforce/scopes/{scopeid}/entries',
|
||||
json=entry, headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()['data']['entryid']
|
||||
|
||||
|
||||
def test_compliance_only_lists_app_linked_entries(client, db, auth_headers):
|
||||
appid = _seed_app(db, 'PC-DMIS')
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'PC-DMIS 2019', 'Type': 'MSI', 'appid': appid,
|
||||
'DetectionMethod': 'FileVersion', 'DetectionPath': 'C:\\p.exe',
|
||||
'DetectionValue': '2019 R1'})
|
||||
# An unlinked entry (no appid) must NOT appear.
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'Unlinked', 'Type': 'MSI', 'Installer': 'apps/x.msi'})
|
||||
|
||||
data = client.get(f'/api/geenforce/scopes/{scopeid}/compliance',
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert data['computersplugin'] is True
|
||||
assert [r['entryname'] for r in data['rows']] == ['PC-DMIS 2019']
|
||||
row = data['rows'][0]
|
||||
assert row['appid'] == appid
|
||||
assert row['appname'] == 'PC-DMIS'
|
||||
assert row['expectedversion'] == '2019 R1'
|
||||
|
||||
|
||||
def test_compliance_counts_active_installs_and_version_match(client, db,
|
||||
auth_headers):
|
||||
appid = _seed_app(db, 'PC-DMIS')
|
||||
# Two PCs on the expected version, one on a different version, one inactive.
|
||||
_seed_pc_with_app(db, 'CMM01', appid, '2019 R1')
|
||||
_seed_pc_with_app(db, 'CMM02', appid, '2019 R1')
|
||||
_seed_pc_with_app(db, 'CMM03', appid, '2016 R2')
|
||||
_seed_pc_with_app(db, 'CMM04', appid, '2019 R1', isactive=False)
|
||||
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'PC-DMIS 2019', 'Type': 'MSI', 'appid': appid,
|
||||
'DetectionMethod': 'FileVersion', 'DetectionPath': 'C:\\p.exe',
|
||||
'DetectionValue': '2019 R1'})
|
||||
|
||||
row = client.get(f'/api/geenforce/scopes/{scopeid}/compliance',
|
||||
headers=auth_headers).get_json()['data']['rows'][0]
|
||||
# Inactive row excluded: 3 active installs, 2 on the expected version.
|
||||
assert row['installedcount'] == 3
|
||||
assert row['versionmatchcount'] == 2
|
||||
assert '2 of 3' in row['coveragenote']
|
||||
|
||||
|
||||
def test_compliance_non_fileversion_has_null_version_target(client, db,
|
||||
auth_headers):
|
||||
appid = _seed_app(db, 'eDNC')
|
||||
_seed_pc_with_app(db, 'DNC01', appid, '5.0')
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
# Registry detection is not a version string -> no expected-version target.
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'eDNC', 'Type': 'MSI', 'appid': appid,
|
||||
'DetectionMethod': 'Registry',
|
||||
'DetectionPath': 'HKLM\\Software\\eDNC'})
|
||||
|
||||
row = client.get(f'/api/geenforce/scopes/{scopeid}/compliance',
|
||||
headers=auth_headers).get_json()['data']['rows'][0]
|
||||
assert row['expectedversion'] is None
|
||||
assert row['installedcount'] == 1
|
||||
assert row['versionmatchcount'] is None
|
||||
assert 'no version target' in row['coveragenote']
|
||||
|
||||
|
||||
def test_compliance_not_installed_anywhere(client, db, auth_headers):
|
||||
appid = _seed_app(db, 'Orphan')
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'Orphan', 'Type': 'MSI', 'appid': appid,
|
||||
'DetectionMethod': 'FileVersion', 'DetectionPath': 'C:\\o.exe',
|
||||
'DetectionValue': '1.0'})
|
||||
|
||||
row = client.get(f'/api/geenforce/scopes/{scopeid}/compliance',
|
||||
headers=auth_headers).get_json()['data']['rows'][0]
|
||||
assert row['installedcount'] == 0
|
||||
assert row['versionmatchcount'] == 0
|
||||
assert row['coveragenote'] == 'not installed on any collected PC'
|
||||
|
||||
|
||||
def test_compliance_scope_not_found(client, db, auth_headers):
|
||||
resp = client.get('/api/geenforce/scopes/999999/compliance',
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_compliance_requires_manage_permission(client, db, member_headers):
|
||||
resp = client.get('/api/geenforce/scopes/1/compliance',
|
||||
headers=member_headers)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_compliance_graceful_without_computers_plugin(client, db, auth_headers,
|
||||
monkeypatch):
|
||||
appid = _seed_app(db, 'PC-DMIS')
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
_add_entry(client, auth_headers, scopeid,
|
||||
{'Name': 'PC-DMIS 2019', 'Type': 'MSI', 'appid': appid,
|
||||
'DetectionMethod': 'FileVersion', 'DetectionPath': 'C:\\p.exe',
|
||||
'DetectionValue': '2019 R1'})
|
||||
monkeypatch.setattr(service, '_installed_app_model', lambda: None)
|
||||
|
||||
data = client.get(f'/api/geenforce/scopes/{scopeid}/compliance',
|
||||
headers=auth_headers).get_json()['data']
|
||||
assert data['computersplugin'] is False
|
||||
row = data['rows'][0]
|
||||
assert row['installedcount'] is None
|
||||
assert row['versionmatchcount'] is None
|
||||
assert 'computers plugin not installed' in row['coveragenote']
|
||||
125
tests/test_plugins/test_geenforce_payload.py
Normal file
125
tests/test_plugins/test_geenforce_payload.py
Normal file
@@ -0,0 +1,125 @@
|
||||
"""GE-Enforce inline payloads: upload small scripts/configs into shopdb.
|
||||
|
||||
An entry can carry one inline payload (<= 1 MB) stored as ManifestPayload bytes.
|
||||
Upload computes a sha256, sets the entry's payloadsource='inline' + payloadref,
|
||||
and enforces one payload row per entry (re-upload replaces). GET returns the
|
||||
raw bytes. Upload needs geenforce.publish; GET needs geenforce.manage. appid and
|
||||
the payload metadata are shopdb-only and never enter the served manifest JSON.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import io
|
||||
|
||||
from plugins.geenforce.models import ManifestPayload
|
||||
|
||||
|
||||
def _create_scope(client, auth_headers, name='gea-shopfloor-cmm'):
|
||||
resp = client.post('/api/geenforce/scopes',
|
||||
json={'scopename': name, 'phase': 'runtime'},
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()['data']['scopeid']
|
||||
|
||||
|
||||
def _add_entry(client, auth_headers, scopeid, name='eDNC install'):
|
||||
resp = client.post(f'/api/geenforce/scopes/{scopeid}/entries',
|
||||
json={'Name': name, 'Type': 'PS1',
|
||||
'Script': 'scripts/x.ps1'},
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
return resp.get_json()['data']['entryid']
|
||||
|
||||
|
||||
def _upload(client, headers, entryid, content, filename='config.reg',
|
||||
content_type='multipart/form-data'):
|
||||
return client.post(
|
||||
f'/api/geenforce/entries/{entryid}/payload',
|
||||
data={'file': (io.BytesIO(content), filename)},
|
||||
content_type=content_type, headers=headers)
|
||||
|
||||
|
||||
def test_upload_stores_payload_and_links_entry(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
content = b'Windows Registry Editor Version 5.00\r\n'
|
||||
|
||||
resp = _upload(client, auth_headers, entryid, content)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
data = resp.get_json()['data']
|
||||
assert data['payloadsource'] == 'inline'
|
||||
assert data['payloadref'] == 'config.reg'
|
||||
assert data['payloadsha256'] == hashlib.sha256(content).hexdigest()
|
||||
assert data['haspayload'] is True
|
||||
|
||||
rows = ManifestPayload.query.filter_by(entryid=entryid).all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].payloadbytes == content
|
||||
|
||||
|
||||
def test_reupload_replaces_single_row(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
_upload(client, auth_headers, entryid, b'first')
|
||||
newcontent = b'second version'
|
||||
resp = _upload(client, auth_headers, entryid, newcontent)
|
||||
assert resp.status_code == 201
|
||||
|
||||
rows = ManifestPayload.query.filter_by(entryid=entryid).all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].payloadbytes == newcontent
|
||||
assert rows[0].payloadsha256 == hashlib.sha256(newcontent).hexdigest()
|
||||
|
||||
|
||||
def test_oversized_payload_rejected(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
toobig = b'x' * (1024 * 1024 + 1)
|
||||
resp = _upload(client, auth_headers, entryid, toobig)
|
||||
assert resp.status_code == 400
|
||||
assert ManifestPayload.query.filter_by(entryid=entryid).count() == 0
|
||||
|
||||
|
||||
def test_empty_and_missing_file_rejected(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
empty = _upload(client, auth_headers, entryid, b'')
|
||||
assert empty.status_code == 400
|
||||
nofile = client.post(f'/api/geenforce/entries/{entryid}/payload',
|
||||
data={}, content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
assert nofile.status_code == 400
|
||||
|
||||
|
||||
def test_download_returns_exact_bytes(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
content = b'\x00\x01binary payload\xff'
|
||||
_upload(client, auth_headers, entryid, content, filename='blob.bin')
|
||||
|
||||
resp = client.get(f'/api/geenforce/entries/{entryid}/payload',
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 200
|
||||
assert resp.data == content
|
||||
assert 'blob.bin' in resp.headers['Content-Disposition']
|
||||
|
||||
|
||||
def test_download_404_when_no_payload(client, db, auth_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
resp = client.get(f'/api/geenforce/entries/{entryid}/payload',
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_upload_404_unknown_entry(client, db, auth_headers):
|
||||
resp = _upload(client, auth_headers, 999999, b'data')
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_upload_requires_publish_permission(client, db, auth_headers,
|
||||
member_headers):
|
||||
scopeid = _create_scope(client, auth_headers)
|
||||
entryid = _add_entry(client, auth_headers, scopeid)
|
||||
# member (no permissions) cannot upload.
|
||||
resp = _upload(client, member_headers, entryid, b'data')
|
||||
assert resp.status_code == 403
|
||||
Reference in New Issue
Block a user