shopdb-flask/plugins/backups/services/ntlars.py

"""NTLARS / DNC registry backup codec.

Converts between Windows .reg files and a dialect-neutral JSON projection.

WHY DIALECT-NEUTRAL: NTLARS is a 32-bit app, so its settings physically live
under HKLM\\SOFTWARE\\WOW6432Node\\GE Aircraft Engines\\DNC. But NTLARS's own
Save... button exports them WITHOUT the WOW6432Node segment (it writes the path
it asks for, before the WOW64 redirector rewrites it). Both dialects therefore
exist in the wild:

  NTLARS Save... output      HKLM\\SOFTWARE\\GE Aircraft Engines\\DNC
  scripted / reg export      HKLM\\SOFTWARE\\WOW6432Node\\GE Aircraft Engines\\DNC

Parsing strips whichever root matched and stores subkeys RELATIVE to it, so the
stored revision commits to neither. render() then re-attaches whichever root the
consumer needs:

  dialect='ntlars'      no WOW6432Node - what the NTLARS Load... button expects
  dialect='wow6432node' explicit - what `reg import` needs on a 64-bit box

Getting this backwards is silent: a reg import of the NTLARS dialect on 64-bit
writes to the 64-bit hive, where NTLARS will never look, and reports success.

CANONICAL ORDERING: keys and value names are sorted on parse. MySQL's JSON type
normalizes object key order anyway, so preserving source order is not possible
end-to-end; sorting makes it deterministic instead, which is what makes diffs
between revisions stable. Re-rendered files are semantically identical to their
source, not byte-identical.
"""

import re

SCHEMA = 'ntlars/1'

REGROOT = 'HKEY_LOCAL_MACHINE'
DNCPATH = r'SOFTWARE\GE Aircraft Engines\DNC'
DNCPATHWOW = r'SOFTWARE\WOW6432Node\GE Aircraft Engines\DNC'

ROOTNTLARS = '{}\\{}'.format(REGROOT, DNCPATH)
ROOTWOW = '{}\\{}'.format(REGROOT, DNCPATHWOW)

# Order is not significant: the two roots diverge immediately after
# 'SOFTWARE\\' (GE vs WOW), so neither is a string prefix of the other and
# _striproot's startswith test cannot match the wrong one. Listed longest-first
# only for readability.
KNOWNROOTS = (ROOTWOW, ROOTNTLARS)

HEADER = 'Windows Registry Editor Version 5.00'

# hex(N): type codes that appear in .reg files, mapped to registry type names.
HEXTYPES = {
    0: 'REG_NONE',
    1: 'REG_SZ',
    2: 'REG_EXPAND_SZ',
    3: 'REG_BINARY',
    4: 'REG_DWORD',
    7: 'REG_MULTI_SZ',
    11: 'REG_QWORD',
}
HEXTYPECODES = {v: k for k, v in HEXTYPES.items()}


class NtlarsParseError(ValueError):
    """Raised when input is not a .reg file we can make sense of."""


def decodereg(raw):
    """Decode .reg bytes to text.

    .reg files are conventionally UTF-16LE with a BOM (that is what both
    regedit and NTLARS emit), but hand-edited ones show up as UTF-8. Sniff the
    BOM rather than trusting the extension.
    """
    if isinstance(raw, str):
        return raw
    if raw.startswith(b'\xff\xfe'):
        return raw.decode('utf-16-le')[1:]
    if raw.startswith(b'\xfe\xff'):
        return raw.decode('utf-16-be')[1:]
    if raw.startswith(b'\xef\xbb\xbf'):
        return raw.decode('utf-8-sig')
    # No BOM. UTF-16LE ASCII text has a NUL in every other byte.
    if b'\x00' in raw[:64]:
        return raw.decode('utf-16-le', errors='replace')
    return raw.decode('utf-8', errors='replace')


def _unescape(s):
    return s.replace('\\\\', '\x00').replace('\\"', '"').replace('\x00', '\\')


def _escape(s):
    return s.replace('\\', '\\\\').replace('"', '\\"')


def _joincontinuations(text):
    """Fold .reg line continuations (trailing backslash) into single lines."""
    out = []
    for line in text.replace('\r\n', '\n').replace('\r', '\n').split('\n'):
        if out and out[-1].endswith('\\'):
            out[-1] = out[-1][:-1] + line.strip()
        else:
            out.append(line)
    return out


def _parsehexvalue(body):
    """Parse the body of a hex:/hex(N): value into (typename, data)."""
    m = re.match(r'^hex(?:\((?P<code>[0-9a-fA-F]+)\))?:(?P<bytes>.*)$', body, re.S)
    if not m:
        raise NtlarsParseError('unparseable hex value: {!r}'.format(body))
    code = int(m.group('code'), 16) if m.group('code') else 3
    tokens = [t.strip() for t in m.group('bytes').split(',') if t.strip()]
    try:
        data = bytes(int(t, 16) for t in tokens)
    except ValueError as exc:
        raise NtlarsParseError('bad hex byte in value: {}'.format(exc))

    typename = HEXTYPES.get(code, 'REG_BINARY')

    # Wide-string hex types decode back to text so diffs stay readable.
    if typename in ('REG_SZ', 'REG_EXPAND_SZ'):
        return typename, data.decode('utf-16-le', errors='replace').rstrip('\x00')
    if typename == 'REG_MULTI_SZ':
        text = data.decode('utf-16-le', errors='replace')
        return typename, [p for p in text.split('\x00') if p]
    # REG_QWORD must come back as an int: _rendervalue turns it back into
    # little-endian bytes via int(), so storing the "aa,bb" byte form here
    # would raise at download time - i.e. precisely when someone is trying to
    # restore a machine.
    if typename == 'REG_QWORD':
        return typename, int.from_bytes(data, 'little')
    return typename, ','.join('{:02x}'.format(b) for b in data)


def _parsevalue(body):
    """Parse the right-hand side of a .reg value assignment."""
    body = body.strip()
    if body.startswith('"'):
        if not body.endswith('"') or len(body) < 2:
            raise NtlarsParseError('unterminated string value: {!r}'.format(body))
        return 'REG_SZ', _unescape(body[1:-1])
    if body.lower().startswith('dword:'):
        try:
            return 'REG_DWORD', int(body.split(':', 1)[1].strip(), 16)
        except ValueError:
            raise NtlarsParseError('bad dword value: {!r}'.format(body))
    if body.lower().startswith('hex'):
        return _parsehexvalue(body)
    if body == '-':
        return 'DELETE', None
    raise NtlarsParseError('unrecognised value form: {!r}'.format(body))


def _striproot(keypath):
    """Strip a known DNC root, returning the relative subkey path.

    Returns None for keys outside the DNC tree so callers can ignore them
    rather than silently folding unrelated hives into the backup.
    """
    for root in KNOWNROOTS:
        if keypath.upper() == root.upper():
            return ''
        prefix = root.upper() + '\\'
        if keypath.upper().startswith(prefix):
            return keypath[len(prefix):]
    return None


def parse(raw):
    """Parse .reg bytes/text into the dialect-neutral JSON projection.

    Returns {'schema', 'sourcedialect', 'keys': [{'path', 'values': {...}}]}
    with keys and value names sorted for deterministic diffing.
    """
    text = decodereg(raw)
    lines = _joincontinuations(text)

    if not any(line.strip().lower().startswith('windows registry editor')
               or line.strip().lower().startswith('regedit4')
               for line in lines[:5]):
        raise NtlarsParseError('missing "Windows Registry Editor" header')

    keys = {}
    current = None
    sawwow = False
    sawplain = False

    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith(';'):
            continue

        if stripped.startswith('[') and stripped.endswith(']'):
            keypath = stripped[1:-1].strip()
            if keypath.startswith('-'):
                current = None          # key deletion, not a backup concern
                continue
            if keypath.upper().startswith(ROOTWOW.upper()):
                sawwow = True
            elif keypath.upper().startswith(ROOTNTLARS.upper()):
                sawplain = True
            rel = _striproot(keypath)
            if rel is None:
                current = None          # outside the DNC tree - ignore
                continue
            current = rel
            keys.setdefault(current, {})
            continue

        if current is None or '=' not in stripped:
            continue

        # Match the QUOTED name and split at the '=' that follows its closing
        # quote. A plain split('=', 1) breaks on any value name containing '='
        # or an escaped quote - both legal in the registry - and silently drops
        # the value.
        match = re.match(r'^(?:@|"((?:[^"\\]|\\.)*)")\s*=\s*(.*)$', stripped, re.S)
        if not match:
            continue
        name = '' if match.group(1) is None else _unescape(match.group(1))
        body = match.group(2)

        # Deliberately NOT caught. A backup that silently dropped an
        # unparseable value would present as complete and restore a machine
        # with a setting missing - the exact silent-failure class this project
        # has repeatedly been bitten by. Fail the whole parse instead; the
        # collector reports it and the previous good revision stays newest.
        typename, data = _parsevalue(body)
        if typename == 'DELETE':
            continue
        keys[current][name] = {'type': typename, 'data': data}

    if not keys:
        raise NtlarsParseError(
            'no keys under {} or {} - not an NTLARS DNC backup'.format(
                ROOTNTLARS, ROOTWOW))

    dialect = 'wow6432node' if sawwow else ('ntlars' if sawplain else 'unknown')

    return {
        'schema': SCHEMA,
        'sourcedialect': dialect,
        'keys': [
            {'path': path, 'values': dict(sorted(keys[path].items()))}
            for path in sorted(keys)
        ],
    }


def _rendervalue(name, entry):
    typename = entry.get('type', 'REG_SZ')
    data = entry.get('data')
    lhs = '@' if name == '' else '"{}"'.format(_escape(name))

    if typename == 'REG_SZ':
        return '{}="{}"'.format(lhs, _escape('' if data is None else str(data)))
    if typename == 'REG_DWORD':
        return '{}=dword:{:08x}'.format(lhs, int(data) & 0xFFFFFFFF)
    if typename == 'REG_QWORD':
        return '{}=hex(b):{}'.format(lhs, _hexbytes(
            int(data).to_bytes(8, 'little')))
    if typename == 'REG_EXPAND_SZ':
        payload = ('' if data is None else str(data)).encode('utf-16-le') + b'\x00\x00'
        return '{}=hex(2):{}'.format(lhs, _hexbytes(payload))
    if typename == 'REG_MULTI_SZ':
        parts = data if isinstance(data, list) else [str(data)]
        payload = ''.join(p + '\x00' for p in parts).encode('utf-16-le') + b'\x00\x00'
        return '{}=hex(7):{}'.format(lhs, _hexbytes(payload))

    # REG_BINARY / REG_NONE: data is the "aa,bb,cc" form parse() produced.
    raw = bytes(int(t, 16) for t in str(data).split(',') if t.strip()) if data else b''
    code = HEXTYPECODES.get(typename, 3)
    prefix = 'hex:' if code == 3 else 'hex({:x}):'.format(code)
    return '{}={}{}'.format(lhs, prefix, _hexbytes(raw))


def _hexbytes(raw):
    return ','.join('{:02x}'.format(b) for b in raw)


def render(projection, dialect='ntlars', comments=None):
    """Render the JSON projection back to .reg bytes (UTF-16LE, BOM, CRLF).

    dialect='ntlars'      omits WOW6432Node - use with the NTLARS Load... button
    dialect='wow6432node' includes it - use with `reg import` on 64-bit
    """
    if dialect not in ('ntlars', 'wow6432node'):
        raise ValueError('unknown dialect: {!r}'.format(dialect))
    root = ROOTWOW if dialect == 'wow6432node' else ROOTNTLARS

    out = [HEADER, '']
    for line in (comments or []):
        out.append('; {}'.format(line))
    if comments:
        out.append('')

    for key in projection.get('keys', []):
        path = key.get('path', '')
        out.append('[{}]'.format(root + ('\\' + path if path else '')))
        for name, entry in (key.get('values') or {}).items():
            out.append(_rendervalue(name, entry))
        out.append('')

    text = '\r\n'.join(out)
    if not text.endswith('\r\n'):
        text += '\r\n'
    return b'\xff\xfe' + text.encode('utf-16-le')