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'

# SECRETS ARE NOT BACKED UP.
#
# The DNC registry carries an FTP password in cleartext. Stored verbatim it
# reached three places it must not: the backuprevisions table, the revision
# panel (which renders values), and the download route (which hands back a
# ready-to-import .reg). Anyone with backups.view could read the fleet's FTP
# password off a web page.
#
# Redaction happens at PARSE, so the secret never enters the database at all.
# Redacting only on display would leave it at rest and one query away.
#
# What is deliberately NOT done: dropping the value silently. This module's
# parser refuses to discard an unparseable value precisely because a backup that
# looks complete and restores a machine with a setting missing is the worst
# outcome. A redacted value keeps its NAME and records itself in
# `redacted`, so both the panel and a restore can say plainly that this one
# field must be set by hand.
REDACTEDMARKER = '***REDACTED***'
SECRETNAMEPATTERN = re.compile(r'passw|pwd|secret|apikey|token', re.IGNORECASE)


def issecretname(name):
    """Whether a registry value name holds a credential."""
    return bool(SECRETNAMEPATTERN.search(name or ''))

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
    if body == '':
        # "Name="  with nothing after the sign. Not strictly legal, but it
        # occurs in real exports (the part marker's WJPRT.reg has KRelay1 like
        # this), and failing the whole file over it would mean that machine can
        # never be backed up. Treat it as an empty string so the value NAME is
        # still preserved.
        return 'REG_SZ', ''
    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')

    # Strip credentials before the projection is returned - i.e. before anything
    # persists it. Recorded by path\name so the omission is visible rather than
    # discovered during a restore.
    redacted = []
    for path in keys:
        for name, value in keys[path].items():
            if not issecretname(name):
                continue
            if value.get('data') in ('', None):
                continue   # nothing to hide
            keys[path][name] = {'type': value.get('type', 'REG_SZ'),
                                'data': REDACTEDMARKER}
            redacted.append('{}\\{}'.format(path, name) if path else name)

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


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

    Redacted credentials are rendered as a comment naming the value, NOT as the
    marker string. Importing '***REDACTED***' as the FTP password would leave a
    machine that looks restored and cannot connect, and the technician would be
    debugging the network. A commented-out value fails loudly at the one moment
    somebody is looking at the file.
    """
    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('')

    redacted = projection.get('redacted') or []
    if redacted:
        out.append('; ---------------------------------------------------------------')
        out.append('; {} credential value(s) were NOT backed up and are NOT in this'
                   .format(len(redacted)))
        out.append('; file. Set them by hand after importing:')
        for name in redacted:
            out.append(';     {}'.format(name))
        out.append('; ---------------------------------------------------------------')
        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():
            # A redacted credential is commented out, never emitted as the
            # marker. Importing the marker as a password produces a machine that
            # looks restored and cannot log in.
            if (entry or {}).get('data') == REDACTEDMARKER:
                out.append('; "{}" was not backed up - set it by hand'.format(
                    _escape(name)))
                continue
            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')