geenforce: display-readiness batch (server hardening, PS client wiring, display scope)

Get GE-Enforce closer to running on credential-less Intune/Entra display PCs
that pull manifest + payloads over HTTPS instead of SMB.

Server (plugins/geenforce/api/routes.py):
- Rate-limit + 512MB served-size ceiling on GET /payload/<sha256> (reuses the
  login limiter's cache pattern, config-overridable via GEENFORCE_PAYLOAD_*).
- New tests: payload hardening, manifestblobs model-vs-migration parity, and a
  report-contract test locking the lowercase per-entry report keys.

PS client (plugins/geenforce/client/):
- Fix New-ShopdbReport per-entry key casing to lowercase (name/action/selfhealed/
  exitcode/message) to match what the server reads; the engine emits PascalCase.
- Enforce TLS 1.2 in the network functions.
- Fetch + merge the fleet-wide common scope alongside the pctype scope
  (pctype wins on conflict; -NoCommon opt-out).
- Normalize whatever the engine returns into a well-formed summary.
- Make the empty-cache fail-safe observable: event-log entry + report ping
  instead of a silent exit 0.

Manifest (plugins/geenforce/seed_display_scope.py + docs/GE-ENFORCE-DISPLAY.md):
- Seed a gea-shopfloor-display scope: 4 Edge kiosk drift-heal registry entries
  + 1 data-driven dispatcher (Dashboard/Lobby/3DPrintRoom via display-type.txt).
  Kiosk EXEs stay image-baked; the manifest heals policy/config drift only.
- Documents the common SMB-payload audit (entries needing http/inline before a
  share-less display can inherit common).

Migration registry (shopdb/plugins/alembic_template.py + test):
- Register the pre-existing manifestblobs and the new printersupplyalerts tables
  in PLUGIN_TABLE_OWNERS; update EXPECTED_HEAD_REVISION for geenforce (0002blobs),
  printers (0002supplyalerts), and printedparts (0004txnrev) which had drifted.
This commit is contained in:
cproudlock
2026-07-23 08:16:38 -04:00
parent b211e817d5
commit 9d65ef103d
12 changed files with 1261 additions and 25 deletions

View File

@@ -46,7 +46,11 @@ CUTOVER_PLUGINS = (
# stamp '<plugin>0001anchor'; measuringtools stamps its real baseline id.
EXPECTED_HEAD_REVISION = {plugin: f'{plugin}0001anchor' for plugin in CUTOVER_PLUGINS}
EXPECTED_HEAD_REVISION['measuringtools'] = 'measuringtools0001baseline'
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0001baseline'
# geenforce adds the content-addressed blob store (manifestblobs) on top of its
# baseline.
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0002blobs'
# printers adds the printersupplyalerts crossing-state table on top of its anchor.
EXPECTED_HEAD_REVISION['printers'] = 'printers0002supplyalerts'
# machines (renamed from equipment) keeps its original anchor id and adds the
# rename revision on top, so its head is not the f-string default.
EXPECTED_HEAD_REVISION['machines'] = 'machines0002rename'
@@ -54,8 +58,9 @@ EXPECTED_HEAD_REVISION['machines'] = 'machines0002rename'
EXPECTED_HEAD_REVISION['employees'] = 'employees0002photo'
# usb drops the dead usbcheckouts.machineid column on top of its anchor.
EXPECTED_HEAD_REVISION['usb'] = 'usb0002dropmachineid'
# printedparts is post-cutover: its 0001 really creates its tables.
EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0003gagetag'
# printedparts is post-cutover: its 0001 really creates its tables; 0004 adds
# the per-transaction revision column.
EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0004txnrev'
# notifications indexes businessunitid on top of its anchor.
EXPECTED_HEAD_REVISION['notifications'] = 'notifications0002buidx'

View File

@@ -0,0 +1,108 @@
"""GE-Enforce manifestblobs model-vs-migration parity gate.
manifestblobs is created by the SECOND geenforce migration
(0002_manifest_blobs.py), and the existing DDL-parity test only diffs the
baseline (0001) tables - it does not even list ManifestBlob. So the http-blob
store table has had NO parity coverage: a column added to the ManifestBlob model
but not the migration (or vice versa) would ship a broken schema silently.
This gate builds manifestblobs two ways and diffs it column-by-column:
1. the ManifestBlob model via metadata.create_all (desired shape), and
2. the 0002 migration's upgrade() run against a fresh engine (shipped shape).
Both go through the same SQLite dialect so reflected types render identically.
"""
import importlib.util
from pathlib import Path
from sqlalchemy import create_engine, inspect
from alembic.migration import MigrationContext
from alembic.operations import Operations
_BLOBS_MIGRATION = (Path(__file__).resolve().parent.parent.parent / 'plugins' /
'geenforce' / 'migrations' / 'versions' /
'0002_manifest_blobs.py')
_EXPECTED_COLUMNS = ('sha256', 'filename', 'contenttype', 'sizebytes',
'createdat')
def _migration_columns():
"""Run the 0002 migration upgrade() on a fresh engine; return the reflected
manifestblobs columns as {name: column-dict}."""
spec = importlib.util.spec_from_file_location('geenforce_blobs',
str(_BLOBS_MIGRATION))
migration = importlib.util.module_from_spec(spec)
spec.loader.exec_module(migration)
engine = create_engine('sqlite://')
with engine.connect() as connection:
operations = Operations(MigrationContext.configure(connection))
operations._install_proxy()
try:
migration.upgrade()
finally:
operations._remove_proxy()
connection.commit()
inspector = inspect(connection)
assert 'manifestblobs' in inspector.get_table_names(), \
'migration did not create manifestblobs'
return {column['name']: column
for column in inspector.get_columns('manifestblobs')}
def _model_columns(db):
"""Build manifestblobs from the ManifestBlob model; return {name: column}."""
from plugins.geenforce.models.manifest import ManifestBlob
engine = create_engine('sqlite://')
db.metadata.create_all(engine, tables=[ManifestBlob.__table__])
return {column['name']: column
for column in inspect(engine).get_columns('manifestblobs')}
def _colkey(column):
"""Normalize a reflected column to the facets we gate on (type, nullable)."""
return (str(column['type']), bool(column['nullable']))
def test_manifestblob_model_matches_migration(app, db):
"""Every ManifestBlob column matches what the 0002 migration builds."""
with app.app_context():
modelcols = _model_columns(db)
migrationcols = _migration_columns()
drift = []
for name in sorted(set(modelcols) - set(migrationcols)):
drift.append(f'{name}: in model, not in migration')
for name in sorted(set(migrationcols) - set(modelcols)):
drift.append(f'{name}: in migration, not in model')
for name in sorted(set(modelcols) & set(migrationcols)):
if _colkey(modelcols[name]) != _colkey(migrationcols[name]):
drift.append(
f'{name}: model {_colkey(modelcols[name])} != '
f'migration {_colkey(migrationcols[name])}')
assert not drift, ('manifestblobs model/migration DDL drift:\n'
+ '\n'.join(drift))
def test_manifestblob_expected_columns_present_both_sides(app, db):
"""Lock the load-bearing columns (sha256 primary key + registry fields)
exist in BOTH the model and the migration."""
with app.app_context():
modelcols = _model_columns(db)
migrationcols = _migration_columns()
for name in _EXPECTED_COLUMNS:
assert name in modelcols, f'{name} missing from ManifestBlob model'
assert name in migrationcols, f'{name} missing from migration'
def test_manifestblob_sha256_is_not_nullable(app, db):
"""The content hash is the primary key: it must be NOT NULL on both sides."""
with app.app_context():
modelcols = _model_columns(db)
migrationcols = _migration_columns()
assert modelcols['sha256']['nullable'] is False
assert migrationcols['sha256']['nullable'] is False

View File

@@ -0,0 +1,108 @@
"""Cover the gea-shopfloor-display authoring seed.
Running seed_display_scope must produce the display scope with the expected
entry count and types: four Registry drift-heal entries (Edge kiosk relaunch
policies) plus one inline PS1 dispatcher. Also checks the dispatcher payload is
stored inline and the display-type map drives the dispatcher script.
"""
from plugins.geenforce.models import (
ManifestScope, ManifestEntry, ManifestPayload,
)
from plugins.geenforce.seed_display_scope import (
seed_display_scope, build_display_manifest, build_dispatcher_script,
DISPLAY_TYPE_TARGETS, SCOPE_NAME, DISPATCHER_FILENAME,
)
def test_seed_creates_display_scope(db):
summary = seed_display_scope()
scope = ManifestScope.query.filter_by(
scopename=SCOPE_NAME, phase='runtime').first()
assert scope is not None
# Not the common scope: the client merges common underneath at fetch time.
assert scope.iscommon is False
# Four Registry drift-heal entries + one PS1 dispatcher, in order.
assert summary['entrycount'] == 5
assert summary['entrytypes'] == ['Registry', 'Registry', 'Registry',
'Registry', 'PS1']
def test_registry_entries_use_valuematches_detection(db):
seed_display_scope()
registry_entries = ManifestEntry.query.filter_by(
entrytype='Registry').order_by(ManifestEntry.sortorder).all()
assert len(registry_entries) == 4
regnames = {entry.regname for entry in registry_entries}
assert regnames == {
'RelaunchNotification', 'RelaunchNotificationPeriod',
'RelaunchHeadsUpPeriod', 'RelaunchWindow',
}
for entry in registry_entries:
assert entry.detectionmethod == 'ValueMatches'
assert entry.regpath == 'HKLM:\\SOFTWARE\\Policies\\Microsoft\\Edge'
# Detection targets the same value it writes, coerced to string.
assert entry.detectionname == entry.regname
assert entry.detectionvalue is not None
def test_dispatcher_is_inline_and_data_driven(db):
summary = seed_display_scope()
dispatcher = ManifestEntry.query.filter_by(entrytype='PS1').one()
assert dispatcher.payloadsource == 'inline'
assert dispatcher.payloadref == DISPATCHER_FILENAME
assert dispatcher.payloadsha256 == summary['dispatchersha256']
payload = ManifestPayload.query.filter_by(
entryid=dispatcher.entryid).one()
scripttext = payload.payloadbytes.decode('utf-8')
# The data-driven map surfaces every display-type and its target route.
for display_type, route in DISPLAY_TYPE_TARGETS.items():
assert display_type in scripttext
assert route in scripttext
assert 'display-type.txt' in scripttext
def test_seed_publish_freezes_a_version(db):
summary = seed_display_scope(publish=True)
assert summary['publishedversion'] == 1
scope = ManifestScope.query.filter_by(
scopename=SCOPE_NAME, phase='runtime').first()
published = scope.publishedversions.filter_by(iscurrent=True).first()
assert published is not None
assert published.versionnumber == 1
assert '"Version": "2.0"' in published.manifestjson
def test_seed_draft_is_idempotent(db):
first = seed_display_scope()
second = seed_display_scope()
# Re-running rebuilds the draft to the same shape (same scope, same count).
assert first['scopeid'] == second['scopeid']
assert first['entrycount'] == second['entrycount']
assert first['dispatchersha256'] == second['dispatchersha256']
entries = ManifestEntry.query.filter_by(scopeid=second['scopeid']).all()
assert len(entries) == 5
def test_build_manifest_has_no_smb_exe_payloads(db):
manifest = build_display_manifest()
# No entry ships an EXE/MSI/File payload from a share: the only payload is
# the inline dispatcher; everything else is a Registry policy heal.
for entry in manifest['Applications']:
assert entry.get('Installer') is None
assert entry.get('Source') is None
if entry['Type'] == 'PS1':
assert entry.get('PayloadSource') == 'inline'
def test_dispatcher_script_is_ascii():
# Plain ASCII only (no smart quotes / em-dashes) so the naming gate stays
# green and the on-PC script parses cleanly.
build_dispatcher_script().encode('ascii')

View File

@@ -0,0 +1,135 @@
"""GE-Enforce payload-download hardening: rate limit + served-size ceiling.
GET /api/geenforce/payload/<sha256> is reachable with only a read-only
geenforce.fetch token, so a leaked display token must not be able to hammer it
or pull unbounded bytes. These tests lock the two bounds added to the route:
a per-IP fixed-window rate limit (same cache extension as the login limiter)
and a size cap that refuses (413) a blob/inline payload over the configured max.
"""
import hashlib
import io
from contextlib import contextmanager
from plugins.geenforce import service
from plugins.geenforce.models import ManifestBlob
def _fetch_key(client, auth_headers):
resp = client.post('/api/apitokens',
json={'name': 'svc', 'scopes': ['geenforce.fetch']},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return {'X-API-Key': resp.get_json()['data']['secret']}
def _store(app, raw, filename='setup.exe'):
with app.app_context():
sha = service.store_blob(raw, filename, 'application/octet-stream')
service.db.session.commit()
return sha
@contextmanager
def _config(app, **overrides):
"""Temporarily override app.config keys (the app fixture is session-scoped,
so every key is restored to its prior value/absence on exit)."""
missing = object()
saved = {key: app.config.get(key, missing) for key in overrides}
app.config.update(overrides)
try:
yield
finally:
for key, value in saved.items():
if value is missing:
app.config.pop(key, None)
else:
app.config[key] = value
def test_blob_within_cap_is_served(client, app, auth_headers):
raw = b'x' * 64
sha = _store(app, raw, 'ok.exe')
headers = _fetch_key(client, auth_headers)
with _config(app, GEENFORCE_PAYLOAD_MAX_BYTES=1024):
resp = client.get(f'/api/geenforce/payload/{sha}', headers=headers)
assert resp.status_code == 200
assert resp.data == raw
def test_oversized_blob_refused_413(client, app, auth_headers):
raw = b'x' * 64
sha = _store(app, raw, 'big.exe')
headers = _fetch_key(client, auth_headers)
with _config(app, GEENFORCE_PAYLOAD_MAX_BYTES=16):
resp = client.get(f'/api/geenforce/payload/{sha}', headers=headers)
assert resp.status_code == 413
# The blob registry row still exists; the cap gates delivery, not storage.
with app.app_context():
assert service.db.session.get(ManifestBlob, sha) is not None
def test_oversized_inline_payload_refused_413(client, app, db, auth_headers):
scopeid = client.post(
'/api/geenforce/scopes',
json={'scopename': 'gea-shopfloor-cmm', 'phase': 'runtime'},
headers=auth_headers).get_json()['data']['scopeid']
entryid = client.post(
f'/api/geenforce/scopes/{scopeid}/entries',
json={'Name': 'eDNC config', 'Type': 'PS1', 'Script': 's.ps1'},
headers=auth_headers).get_json()['data']['entryid']
content = b'inline config bytes here'
up = client.post(f'/api/geenforce/entries/{entryid}/payload',
data={'file': (io.BytesIO(content), 'config.reg')},
content_type='multipart/form-data', headers=auth_headers)
assert up.status_code == 201, up.get_json()
sha = hashlib.sha256(content).hexdigest()
headers = _fetch_key(client, auth_headers)
with _config(app, GEENFORCE_PAYLOAD_MAX_BYTES=4):
resp = client.get(f'/api/geenforce/payload/{sha}', headers=headers)
assert resp.status_code == 413
def test_payload_download_is_rate_limited(client, app, auth_headers):
sha = _store(app, b'small blob')
key = _fetch_key(client, auth_headers)
# A unique caller IP isolates this bucket from other tests' shared counter.
headers = {**key, 'X-Forwarded-For': '203.0.113.201'}
with _config(app, GEENFORCE_PAYLOAD_RATELIMIT_MAX=3,
GEENFORCE_PAYLOAD_RATELIMIT_WINDOW_SECONDS=300):
for _ in range(3):
ok = client.get(f'/api/geenforce/payload/{sha}', headers=headers)
assert ok.status_code == 200, ok.get_json()
# The 4th request in the same window is over budget.
over = client.get(f'/api/geenforce/payload/{sha}', headers=headers)
assert over.status_code == 429
def test_rate_limit_is_per_ip(client, app, auth_headers):
"""A second caller (different X-Forwarded-For) has its own budget: one IP
being throttled must not throttle everyone."""
sha = _store(app, b'per-ip blob')
key = _fetch_key(client, auth_headers)
hot = {**key, 'X-Forwarded-For': '203.0.113.202'}
cool = {**key, 'X-Forwarded-For': '203.0.113.203'}
with _config(app, GEENFORCE_PAYLOAD_RATELIMIT_MAX=1,
GEENFORCE_PAYLOAD_RATELIMIT_WINDOW_SECONDS=300):
assert client.get(f'/api/geenforce/payload/{sha}',
headers=hot).status_code == 200
assert client.get(f'/api/geenforce/payload/{sha}',
headers=hot).status_code == 429
# Fresh IP still gets its first request through.
assert client.get(f'/api/geenforce/payload/{sha}',
headers=cool).status_code == 200
def test_rate_limit_can_be_disabled(client, app, auth_headers):
sha = _store(app, b'unthrottled blob')
key = _fetch_key(client, auth_headers)
headers = {**key, 'X-Forwarded-For': '203.0.113.204'}
with _config(app, GEENFORCE_PAYLOAD_RATELIMIT_ENABLED=False,
GEENFORCE_PAYLOAD_RATELIMIT_MAX=1):
for _ in range(5):
assert client.get(f'/api/geenforce/payload/{sha}',
headers=headers).status_code == 200

View File

@@ -0,0 +1,137 @@
"""GE-Enforce enforcement-report per-entry key contract.
The server reads LOWERCASE keys from each results[] item -
name / action / selfhealed / exitcode / message (see
service.record_enforcement_report) - and maps them onto the
ManifestEnforcementResult columns entryname / action / selfhealed / exitcode /
message. This is the exact contract the PowerShell client fix targets: the
client emits lowercase keys, so if the server ever silently switched to reading
uppercase keys the client's entryname/selfhealed would drop to empty/false.
These tests POST a realistic multi-entry cycle with the lowercase keys and lock
that the result columns populate, plus the negative case (uppercase keys do NOT
populate) so a future contract flip fails loudly here.
"""
from plugins.geenforce import service
SCOPE = {
'Version': '2.6',
'Applications': [
{'Name': 'VNC firewall rule', 'Type': 'PS1', 'Script': 'scripts/vnc.ps1'},
{'Name': 'PC-DMIS 2023.1', 'Type': 'MSI', 'Installer': 'apps/pcdmis.msi'},
{'Name': 'eDNC config', 'Type': 'PS1', 'Script': 'scripts/ednc.ps1'},
{'Name': 'Blancco agent', 'Type': 'MSI', 'Installer': 'apps/blancco.msi'},
],
}
def _seed_and_publish(app, scopename='gea-shopfloor-cmm'):
with app.app_context():
service.replace_scope_draft(scopename, 'runtime', SCOPE)
service.publish_scope(scopename, 'runtime', notes='v')
service.db.session.commit()
def _report_token(client, auth_headers):
resp = client.post('/api/apitokens',
json={'name': 'svc', 'scopes': ['geenforce.report']},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return resp.get_json()['data']['secret']
def test_lowercase_keys_populate_result_columns(client, db, app, auth_headers):
_seed_and_publish(app)
secret = _report_token(client, auth_headers)
post = client.post('/api/geenforce/report', json={
'hostname': 'WJDISPLAY07',
'scopename': 'gea-shopfloor-cmm',
'phase': 'runtime',
'appliedversion': 1,
'enforcerversion': '2.6',
'counts': {'installed': 2, 'skipped': 1, 'failed': 1, 'filtered': 0},
'results': [
{'name': 'VNC firewall rule', 'action': 'installed',
'selfhealed': True, 'exitcode': 0, 'message': 'rule re-added'},
{'name': 'PC-DMIS 2023.1', 'action': 'skipped',
'selfhealed': False, 'exitcode': 0, 'message': ''},
{'name': 'eDNC config', 'action': 'installed',
'selfhealed': True, 'exitcode': 0},
{'name': 'Blancco agent', 'action': 'failed',
'selfhealed': False, 'exitcode': 1603, 'message': 'MSI 1603'},
],
}, headers={'X-API-Key': secret})
assert post.status_code == 200, post.get_json()
reportid = post.get_json()['data']['reportid']
detail = client.get(f'/api/geenforce/reports/{reportid}',
headers=auth_headers).get_json()['data']
byname = {r['entryname']: r for r in detail['results']}
# Every lowercase 'name' mapped onto entryname (nothing dropped).
assert set(byname) == {'VNC firewall rule', 'PC-DMIS 2023.1',
'eDNC config', 'Blancco agent'}
vnc = byname['VNC firewall rule']
assert vnc['action'] == 'installed'
assert vnc['selfhealed'] is True
assert vnc['exitcode'] == 0
assert vnc['message'] == 'rule re-added'
assert byname['PC-DMIS 2023.1']['action'] == 'skipped'
assert byname['PC-DMIS 2023.1']['selfhealed'] is False
ednc = byname['eDNC config']
assert ednc['selfhealed'] is True
assert ednc['message'] is None # omitted key -> None
blancco = byname['Blancco agent']
assert blancco['action'] == 'failed'
assert blancco['exitcode'] == 1603
assert blancco['message'] == 'MSI 1603'
# A failure is present, so the cycle status is 'failed' (a self-heal without
# any failure would be 'selfhealed').
assert detail['status'] == 'failed'
def test_selfhealed_defaults_false_when_key_omitted(client, db, app,
auth_headers):
_seed_and_publish(app)
secret = _report_token(client, auth_headers)
post = client.post('/api/geenforce/report', json={
'hostname': 'WJDISPLAY08',
'scopename': 'gea-shopfloor-cmm',
'results': [{'name': 'asset report', 'action': 'installed'}],
}, headers={'X-API-Key': secret})
reportid = post.get_json()['data']['reportid']
detail = client.get(f'/api/geenforce/reports/{reportid}',
headers=auth_headers).get_json()['data']
result = detail['results'][0]
assert result['entryname'] == 'asset report'
assert result['selfhealed'] is False
def test_uppercase_keys_do_not_populate(client, db, app, auth_headers):
"""Negative lock: the server reads LOWERCASE keys, so an uppercase-keyed
item does NOT populate entryname/selfhealed. Guards against a silent flip to
uppercase that would break the client's lowercase payload."""
_seed_and_publish(app)
secret = _report_token(client, auth_headers)
post = client.post('/api/geenforce/report', json={
'hostname': 'WJDISPLAY09',
'scopename': 'gea-shopfloor-cmm',
'results': [{'Name': 'wrong casing', 'Action': 'installed',
'SelfHealed': True}],
}, headers={'X-API-Key': secret})
reportid = post.get_json()['data']['reportid']
detail = client.get(f'/api/geenforce/reports/{reportid}',
headers=auth_headers).get_json()['data']
result = detail['results'][0]
assert result['entryname'] == '' # 'Name' not read -> default ''
assert result['selfhealed'] is False # 'SelfHealed' not read -> default