shopdb-flask/tests/test_plugins/test_geenforce_manifest.py

"""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_spoofed_forwarded_for_does_not_bypass_allowlist(client, db, app):
    # SECURITY: the allowlist uses remote_addr, not X-Forwarded-For. A caller
    # whose real IP (127.0.0.1) is NOT allowlisted must NOT gain token-less access
    # by forging X-Forwarded-For to an allowlisted address.
    _seed_and_publish(app)
    _set_allowlist(app, '10.134.48.0/23')  # test client 127.0.0.1 is NOT in it
    resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                      headers={'X-Forwarded-For': '10.134.48.10'})
    assert resp.status_code == 401, 'spoofed X-Forwarded-For bypassed the allowlist'


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']