shopdb-flask/plugins/geenforce/serializer.py

"""Rebuild manifest JSON from DB rows, and canonicalize entries for parity.

`scope_to_manifest` renders a ManifestScope back into the exact JSON structure
GE-Enforce consumes (Version / _comment / Applications, plus Site for
preinstall). `canonical_entry` reduces an Applications[] entry (from either the
on-share file or a rebuilt scope) to only the fields the engine reads, dropping
`_comment` and key order, so the parity harness can compare behavior not bytes.
"""

import json

# Scalar entry fields: model attribute -> manifest key. Emitted only when set,
# so a rebuilt entry has the same sparse key presence as the original.
_SCALAR_FIELDS = [
    ('name', 'Name'),
    ('entrytype', 'Type'),
    ('installer', 'Installer'),
    ('installargs', 'InstallArgs'),
    ('scriptpath', 'Script'),
    ('scriptargs', 'Args'),
    ('sourcepath', 'Source'),
    ('destination', 'Destination'),
    ('regpath', 'RegPath'),
    ('regname', 'RegName'),
    ('regtype', 'RegType'),
    ('detectionmethod', 'DetectionMethod'),
    ('detectionpath', 'DetectionPath'),
    ('detectionname', 'DetectionName'),
    ('detectionvalue', 'DetectionValue'),
    ('detectionpattern', 'DetectionPattern'),
    ('cmmversion', '_CmmVersion'),
    ('logfile', 'LogFile'),
    ('waittimeoutsec', 'WaitTimeoutSec'),
    ('applymode', 'ApplyMode'),
    ('updatewindow', 'UpdateWindow'),
]

# Boolean flags emitted only when True (matches how preinstall.json carries them).
_BOOL_FLAGS = [
    ('preenrollment', 'PreEnrollment'),
    ('killafterdetection', 'KillAfterDetection'),
    ('pctypesstrict', 'PCTypesStrict'),
]

# The complete set of engine-relevant keys the parity check compares.
ENGINE_KEYS = (
    [mk for _, mk in _SCALAR_FIELDS]
    + [mk for _, mk in _BOOL_FLAGS]
    + ['PCTypes', 'TargetHostnames', 'TargetMachineNumbers', 'RegValue', 'InUseCheck']
)


def entry_to_dict(entry):
    """Render a ManifestEntry model back into a manifest Applications[] entry."""
    result = {}
    if entry.comment:
        result['_comment'] = entry.comment
    for attr, key in _SCALAR_FIELDS:
        value = getattr(entry, attr)
        if value is not None and value != '':
            result[key] = value
    # RegValue: stored as a raw JSON literal, reconstitute its real type.
    if entry.regvalue is not None:
        result['RegValue'] = json.loads(entry.regvalue)
    # Multi-value filters (only when present).
    if entry.pctypes:
        result['PCTypes'] = [p.pctypevalue for p in entry.pctypes]
    if entry.hostnames:
        result['TargetHostnames'] = [h.hostnamepattern for h in entry.hostnames]
    if entry.machinenumbers:
        result['TargetMachineNumbers'] = [m.machinenumber
                                          for m in entry.machinenumbers]
    for attr, key in _BOOL_FLAGS:
        if getattr(entry, attr):
            result[key] = True
    # Payload transport for share-less (http/inline) delivery. Omitted for the
    # default 'smb' source so existing share manifests round-trip unchanged; the
    # client fetches GET /api/geenforce/payload/<PayloadSha256> for http/inline.
    if entry.payloadsource and entry.payloadsource != 'smb':
        result['PayloadSource'] = entry.payloadsource
        if entry.payloadsha256:
            result['PayloadSha256'] = entry.payloadsha256
        if entry.payloadref:
            result['PayloadRef'] = entry.payloadref
    # InUseCheck (nested object + Processes[]).
    if entry.inusecheck:
        procs = []
        for proc in entry.inusecheck.processes:
            pd = {'Name': proc.processname}
            if proc.exepath is not None:
                pd['ExePath'] = proc.exepath
            if proc.gracefulclosetimeoutsec is not None:
                pd['GracefulCloseTimeoutSec'] = proc.gracefulclosetimeoutsec
            procs.append(pd)
        result['InUseCheck'] = {
            'Behavior': entry.inusecheck.behavior,
            'Processes': procs,
        }
    return result


def scope_to_manifest(scope):
    """Render a ManifestScope back into a full manifest.json structure."""
    manifest = {'Version': scope.manifestversion}
    if scope.topcomment:
        manifest['_comment'] = scope.topcomment
    if scope.site:
        manifest['Site'] = scope.site
    manifest['Applications'] = [entry_to_dict(e) for e in scope.entries]
    return manifest


def scope_to_json(scope, indent=2):
    """The serialized JSON text a client/export would receive."""
    return json.dumps(scope_to_manifest(scope), indent=indent)


def canonical_entry(entry_dict):
    """Reduce an Applications[] entry to engine-relevant fields for parity.

    Drops `_comment` and key order. Normalizes InUseCheck and the process list
    so two entries that behave identically compare equal regardless of source.
    """
    canon = {}
    for key in ENGINE_KEYS:
        if key == 'InUseCheck':
            continue
        if key in entry_dict and entry_dict[key] is not None and entry_dict[key] != '':
            canon[key] = entry_dict[key]
    inuse = entry_dict.get('InUseCheck')
    if inuse:
        procs = []
        for proc in inuse.get('Processes', []):
            pd = {'Name': proc.get('Name')}
            if proc.get('ExePath') is not None:
                pd['ExePath'] = proc.get('ExePath')
            if proc.get('GracefulCloseTimeoutSec') is not None:
                pd['GracefulCloseTimeoutSec'] = proc.get('GracefulCloseTimeoutSec')
            procs.append(pd)
        canon['InUseCheck'] = {'Behavior': inuse.get('Behavior'),
                               'Processes': procs}
    return canon