Cam Proudlock
a9d0dfff4a
Flask + Vue rewrite of the classic-ASP shopdb, built as a plugin framework so sister GE Aerospace sites can adopt it. Product version 0.7.0-pre-0.8.0, plugin contract 0.11.0, 966 backend tests. Key entry points: - docs/INSTALL-WINDOWS-IIS.md - tested Windows + IIS install runbook (own-site and subpath deployment methods) - docs/PILOT-DEPLOY.md - production pilot: stand up + import + verify - docs/IMPORT-API.md - schema-agnostic legacy data import contract - scripts/site_imports/wjf/ - West Jefferson reference import loader - docs/adr/ - twelve accepted architecture decision records
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
|