4e5b4228c1
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>
126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
"""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
|