Fix defects found in session review of GE-Enforce plugin

Consolidated fixes from a three-dimension adversarial review.

Data-loss (HIGH): the manifest entry editor stripped fields the form did not
expose, because PUT /entries is a full reset-then-apply. The form now captures
everything - InUseCheck processes as structured name/ExePath/timeout rows (not
just names), LogFile, and the three preinstall flags as checkboxes; the dead
payload-source control (never wired) is removed. New regression test proves an
edit preserves ExePath/timeout/LogFile/PreEnrollment/PCTypesStrict.

Update-entry crash (found by that regression test): replacing an entry's
one-to-one InUseCheck (unique entryid) collided with the old row mid-flush ->
IntegrityError -> 400. update_entry now frees the old InUseCheck (delete+flush)
before populate re-inserts it.

Export truncation (MEDIUM): export_scope_to_share used a plain truncating open,
so a failed/partial write left the live on-share manifest (every PC reads it)
empty. Now writes a temp file in the same dir and os.replace() atomically.

Report dedup case bug (MEDIUM, confirmed by scratch test): the iscurrent demote
matched hostname case-sensitively while the read path uses ilike, so a PC
reporting different casing left two iscurrent rows and double-counted. Demote is
now case-insensitive; regression test added.

Simulator fidelity (MEDIUM): PCTypesStrict was captured but ignored by the
filter mirror, so the simulator wrongly matched a collections-only strict entry
to a nocollections PC via the shared Standard alias group. matches_pctype now
honors PCTypesStrict (disables alias expansion); test added.

Hardening: removed the dead/unscoped GEENFORCE_API_KEY env fallback (never wired
into config; tokens are the only path); create/update entry return 400 on a
duplicate Name instead of 500; parity now asserts scope-level Version/Site; a
new test guards real-manifest field lengths against column limits (the DB-free
parity harness can't see truncation); error handling added to the previously
unguarded editor + reports API calls.

Full suite green; naming + frontend build green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

plugins/geenforce/api/routes.py

@@ -11,8 +11,9 @@ Two audiences:
 
 from functools import wraps
 
-from flask import Blueprint, request, current_app, Response
+from flask import Blueprint, request, Response
 from flask_jwt_extended import jwt_required
+from sqlalchemy.exc import IntegrityError
 
 from shopdb.api import (
     db, success_response, error_response, ErrorCodes, require_permission,
@@ -40,15 +41,17 @@ REPORT_SCOPE = 'geenforce.report'
 
 
 def _require_service_token(scope):
-    """Decorator factory: require `scope` service token OR the env bootstrap key."""
+    """Decorator factory: require a managed service token scoped for `scope`.
+
+    Tokens are the only client auth path (the client kit provisions a
+    geenforce.fetch/report token). No env-key fallback: it was never wired into
+    config and an unscoped shared key is a needless backdoor.
+    """
     def wrapper(f):
         @wraps(f)
         def decorated(*args, **kwargs):
             if service_token_authorized(scope):
                 return f(*args, **kwargs)
-            expected = current_app.config.get('GEENFORCE_API_KEY')
-            if expected and request.headers.get('X-API-Key') == expected:
-                return f(*args, **kwargs)
             return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
                                   http_code=401)
         return decorated
@@ -278,7 +281,13 @@ def create_entry(scopeid):
     nextorder = max([e.sortorder for e in scope.entries], default=-1) + 1
     entry = build_entry(payload, nextorder)
     scope.entries.append(entry)
-    db.session.commit()
+    try:
+        db.session.commit()
+    except IntegrityError:
+        db.session.rollback()
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'an entry with that Name already exists in this scope',
+                              http_code=400)
     return success_response(_entry_payload(entry), http_code=201)
 
 
@@ -293,8 +302,20 @@ def update_entry(entryid):
     invalid = _validate_entry(payload)
     if invalid:
         return error_response(ErrorCodes.VALIDATION_ERROR, invalid, http_code=400)
+    # Free the one-to-one InUseCheck (unique entryid) before populate re-inserts
+    # it, so the replacement does not collide with the old row mid-flush.
+    if entry.inusecheck is not None:
+        db.session.delete(entry.inusecheck)
+        entry.inusecheck = None
+        db.session.flush()
     populate_entry(entry, payload)
-    db.session.commit()
+    try:
+        db.session.commit()
+    except IntegrityError:
+        db.session.rollback()
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'an entry with that Name already exists in this scope',
+                              http_code=400)
     return success_response(_entry_payload(entry))
 
 

plugins/geenforce/filters.py

@@ -43,13 +43,22 @@ def _alias_sets(name):
 
 
 def matches_pctype(entry, pctype, subtype=None):
-    """Test-PCTypeMatches: no PCTypes = all; '*' = all; alias-set intersection."""
+    """Test-PCTypeMatches: no PCTypes = all; '*' = all; alias-set intersection.
+
+    PCTypesStrict=true disables alias expansion (both the PC's identity and the
+    manifest values are compared literally), so e.g. a 'collections'-only strict
+    entry does NOT match a 'nocollections' PC via their shared 'Standard' alias
+    group. The preinstall runner honors this flag (see preinstall.json UDC entry);
+    mirroring it keeps the "what would this PC get" simulator correct.
+    """
     values = entry.get('PCTypes') or []
     if not values:
         return True
     if not pctype:
         return True
-    # Names the current PC matches: bare type, "type-subtype", and all aliases.
+    strict = bool(entry.get('PCTypesStrict'))
+    # Names the current PC matches: bare type, "type-subtype", and (unless strict)
+    # all aliases of either.
     mynames = set()
     mynames.add(pctype.lower())
     seeds = [pctype]
@@ -57,15 +66,18 @@ def matches_pctype(entry, pctype, subtype=None):
         combined = f'{pctype}-{subtype}'
         mynames.add(combined.lower())
         seeds.append(combined)
-    for seed in seeds:
-        for group in _alias_sets(seed):
-            for alias in group:
-                mynames.add(alias.lower())
+    if not strict:
+        for seed in seeds:
+            for group in _alias_sets(seed):
+                for alias in group:
+                    mynames.add(alias.lower())
     for value in values:
         if value == '*':
             return True
         if value.lower() in mynames:
             return True
+        if strict:
+            continue
         # The manifest value may itself be an alias - expand and check overlap.
         for group in _alias_sets(value):
             for alias in group:

plugins/geenforce/parity.py

@@ -34,10 +34,19 @@ def check_scope(scopename, phase, original, fixtures):
     orig_apps = original.get('Applications') or []
     rebuilt_apps = rebuilt.get('Applications') or []
 
+    # Check 0: scope-level fields (Version, Site) round-trip.
+    firstdiff = None
+    for key in ('Version', 'Site'):
+        if str(original.get(key) or '') != str(rebuilt.get(key) or ''):
+            firstdiff = (f'scope field {key}: '
+                         f'{original.get(key)!r} vs {rebuilt.get(key)!r}')
+            break
+
+    scope_ok = firstdiff is None
+
     # Check 1: field-identical, order-preserving.
     identical = 0
-    firstdiff = None
-    if len(orig_apps) != len(rebuilt_apps):
+    if firstdiff is None and len(orig_apps) != len(rebuilt_apps):
         firstdiff = (f'entry count {len(orig_apps)} vs {len(rebuilt_apps)}')
     for i in range(min(len(orig_apps), len(rebuilt_apps))):
         co = canonical_entry(orig_apps[i])
@@ -59,7 +68,8 @@ def check_scope(scopename, phase, original, fixtures):
         elif firstdiff is None:
             firstdiff = f'filter mismatch for profile {profile.get("label")}'
 
-    passed = (identical == len(orig_apps) == len(rebuilt_apps)
+    passed = (scope_ok
+              and identical == len(orig_apps) == len(rebuilt_apps)
               and profiles_same == len(fixtures))
     return {
         'scopename': scopename,

plugins/geenforce/service.py

@@ -8,6 +8,7 @@ Kept out of the CLI and routes so both share one implementation:
 """
 
 import os
+import tempfile
 from datetime import datetime, timezone
 
 from sqlalchemy import func
@@ -122,10 +123,15 @@ def record_enforcement_report(payload):
     else:
         status = 'ok'
 
-    # Demote the prior current report for this host+scope.
-    ManifestEnforcementReport.query.filter_by(
-        hostname=hostname, scopename=scopename, phase=phase, iscurrent=True
-    ).update({'iscurrent': False})
+    # Demote the prior current report for this host+scope. Hostname match is
+    # case-insensitive to match the ilike read path - otherwise a PC reporting
+    # its name in different casing would leave two iscurrent rows.
+    ManifestEnforcementReport.query.filter(
+        func.lower(ManifestEnforcementReport.hostname) == hostname.lower(),
+        ManifestEnforcementReport.scopename == scopename,
+        ManifestEnforcementReport.phase == phase,
+        ManifestEnforcementReport.iscurrent == True
+    ).update({'iscurrent': False}, synchronize_session=False)
 
     report = ManifestEnforcementReport(
         hostname=hostname,
@@ -189,6 +195,17 @@ def export_scope_to_share(scopename, phase, shareroot):
         with open(os.path.join(historydir, f'{stamp}-{scopename}.json'), 'w') as dst:
             dst.write(old)
 
-    with open(target, 'w') as handle:
-        handle.write(published.manifestjson)
+    # Atomic write: a partial/failed write must never leave the live on-share
+    # manifest (which every PC reads) truncated. Write a temp file in the same
+    # directory, then rename over the target.
+    targetdir = os.path.dirname(target)
+    fd, tmppath = tempfile.mkstemp(dir=targetdir, suffix='.tmp')
+    try:
+        with os.fdopen(fd, 'w') as handle:
+            handle.write(published.manifestjson)
+        os.replace(tmppath, target)
+    except Exception:
+        if os.path.exists(tmppath):
+            os.remove(tmppath)
+        raise
     return target