0860aa85c5
Fleet PCs on a trusted (vaulted) network can now reach the GE-Enforce client endpoints (manifest, payload, report) without a per-PC token: the auth path accepts a valid geenforce.fetch/report token OR a source IP in the configured allowlist (setting geenforce_allowed_cidrs). Fail-closed; an empty allowlist means the token stays the only path, so existing deployments are unchanged. Rationale: the client token lives in HKLM on every kiosk, so it does not defend against a compromised kiosk anyway - network-perimeter trust is the same practical strength with far less provisioning + no token-rotation churn on a DB wipe. Documented in-UI that this is perimeter trust, not per-device identity. - _ip_allowlisted() (ipaddress, X-Forwarded-For-aware via _client_ip) - /geenforce/config GET/PUT extended with allowedcidrs, server-validated + normalized (bad CIDR -> 400) - new GE-Enforce > Settings tab (GeEnforceSettings.vue) to edit the allowlist in admin, no SQL - 3 regression tests (allow by IP, reject outside list, empty = token required)
185 lines
7.4 KiB
Python
185 lines
7.4 KiB
Python
"""GE-Enforce first slice: import a scope, publish it, serve it to a client.
|
|
|
|
Exercises the vertical the plan calls the first slice (via a small synthetic
|
|
scope rather than the real gea-shopfloor-cmm): draft import -> publish ->
|
|
GET /api/geenforce/manifest with a geenforce.fetch service token; ETag/304;
|
|
draft edits never change the served bytes; publish + rollback change them.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from plugins.geenforce.models import ManifestScope
|
|
from plugins.geenforce import service
|
|
|
|
|
|
SCOPE = {
|
|
'Version': '2.6',
|
|
'Applications': [
|
|
{'Name': 'Alpha', 'Type': 'MSI', 'Installer': 'apps/alpha.msi',
|
|
'DetectionMethod': 'Registry', 'DetectionPath': 'HKLM:\\SOFTWARE\\Alpha'},
|
|
{'Name': 'Beta', 'Type': 'PS1', 'Script': 'scripts/beta.ps1',
|
|
'DetectionMethod': 'Always'},
|
|
],
|
|
}
|
|
|
|
|
|
def _seed_and_publish(app, scopename='gea-shopfloor-cmm', manifest=None):
|
|
with app.app_context():
|
|
service.replace_scope_draft(scopename, 'runtime', manifest or SCOPE)
|
|
service.publish_scope(scopename, 'runtime', notes='initial')
|
|
service.db.session.commit()
|
|
|
|
|
|
def _mint_fetch_token(client, auth_headers):
|
|
resp = client.post('/api/apitokens',
|
|
json={'name': 'geenforce svc', 'scopes': ['geenforce.fetch']},
|
|
headers=auth_headers)
|
|
assert resp.status_code == 201, resp.get_json()
|
|
return resp.get_json()['data']['secret']
|
|
|
|
|
|
def test_import_publish_serve(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
secret = _mint_fetch_token(client, auth_headers)
|
|
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret})
|
|
assert resp.status_code == 200, resp.data
|
|
body = json.loads(resp.data)
|
|
assert [e['Name'] for e in body['Applications']] == ['Alpha', 'Beta']
|
|
assert resp.headers.get('ETag')
|
|
assert resp.headers.get('X-Manifest-Version') == '1'
|
|
|
|
|
|
def test_etag_304(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
secret = _mint_fetch_token(client, auth_headers)
|
|
first = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret})
|
|
etag = first.headers['ETag']
|
|
again = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret, 'If-None-Match': etag})
|
|
assert again.status_code == 304
|
|
|
|
|
|
def test_draft_edit_does_not_change_served_bytes(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
secret = _mint_fetch_token(client, auth_headers)
|
|
before = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret}).data
|
|
|
|
# Edit the DRAFT (add an entry) but do NOT publish.
|
|
changed = {'Version': '2.6', 'Applications': SCOPE['Applications'] + [
|
|
{'Name': 'Gamma', 'Type': 'PS1', 'Script': 'scripts/gamma.ps1'}]}
|
|
with app.app_context():
|
|
service.replace_scope_draft('gea-shopfloor-cmm', 'runtime', changed)
|
|
service.db.session.commit()
|
|
|
|
after = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret}).data
|
|
assert before == after # served bytes come from the published snapshot only
|
|
|
|
|
|
def test_publish_then_rollback(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
secret = _mint_fetch_token(client, auth_headers)
|
|
v1 = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret}).data
|
|
|
|
changed = {'Version': '2.6', 'Applications': SCOPE['Applications'] + [
|
|
{'Name': 'Gamma', 'Type': 'PS1', 'Script': 'scripts/gamma.ps1'}]}
|
|
with app.app_context():
|
|
service.replace_scope_draft('gea-shopfloor-cmm', 'runtime', changed)
|
|
service.publish_scope('gea-shopfloor-cmm', 'runtime', notes='add gamma')
|
|
service.db.session.commit()
|
|
|
|
v2 = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret})
|
|
assert v2.headers['X-Manifest-Version'] == '2'
|
|
assert b'Gamma' in v2.data
|
|
|
|
with app.app_context():
|
|
service.rollback_scope('gea-shopfloor-cmm', 'runtime', 1)
|
|
service.db.session.commit()
|
|
|
|
rolled = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret})
|
|
assert rolled.data == v1 # byte-identical: published snapshots are frozen text
|
|
|
|
|
|
def test_unauthenticated_rejected(client, db, app):
|
|
_seed_and_publish(app)
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def _set_allowlist(app, value):
|
|
from shopdb.api import Setting
|
|
with app.app_context():
|
|
Setting.set('geenforce_allowed_cidrs', value, 'string', 'geenforce')
|
|
service.db.session.commit()
|
|
|
|
|
|
def test_ip_allowlist_allows_without_token(client, db, app):
|
|
# An allowlisted caller reaches the manifest with NO token (vault trust).
|
|
_seed_and_publish(app)
|
|
_set_allowlist(app, '127.0.0.0/8, 10.134.48.0/23') # test client is 127.0.0.1
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
|
|
assert resp.status_code == 200, resp.data
|
|
assert b'Alpha' in resp.data
|
|
|
|
|
|
def test_ip_not_in_allowlist_still_rejected(client, db, app):
|
|
# A caller outside the allowlist and with no token is refused (fail-closed).
|
|
_seed_and_publish(app)
|
|
_set_allowlist(app, '10.0.0.0/8') # test client 127.0.0.1 is NOT in 10/8
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_empty_allowlist_keeps_token_required(client, db, app):
|
|
# Empty/unset allowlist = disabled; token stays the only path (back-compat).
|
|
_seed_and_publish(app)
|
|
_set_allowlist(app, '')
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_wrong_scope_rejected(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
resp = client.post('/api/apitokens',
|
|
json={'name': 'wrong', 'scopes': ['collector.ingest']},
|
|
headers=auth_headers)
|
|
secret = resp.get_json()['data']['secret']
|
|
manifest = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
|
headers={'X-API-Key': secret})
|
|
assert manifest.status_code == 401
|
|
|
|
|
|
def test_unpublished_scope_404(client, db, app, auth_headers):
|
|
with app.app_context():
|
|
service.replace_scope_draft('gea-shopfloor-keyence', 'runtime', SCOPE)
|
|
service.db.session.commit() # draft only, never published
|
|
secret = _mint_fetch_token(client, auth_headers)
|
|
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-keyence',
|
|
headers={'X-API-Key': secret})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_admin_list_and_preview(client, db, app, auth_headers):
|
|
_seed_and_publish(app)
|
|
listing = client.get('/api/geenforce/scopes', headers=auth_headers)
|
|
assert listing.status_code == 200
|
|
scopes = listing.get_json()['data']
|
|
cmm = next(s for s in scopes if s['scopename'] == 'gea-shopfloor-cmm')
|
|
assert cmm['entrycount'] == 2
|
|
assert cmm['publishedversion'] == 1
|
|
|
|
preview = client.get(f"/api/geenforce/scopes/{cmm['scopeid']}/preview",
|
|
headers=auth_headers)
|
|
assert preview.status_code == 200
|
|
assert [e['Name'] for e in preview.get_json()['data']['manifest']['Applications']] \
|
|
== ['Alpha', 'Beta']
|