backups plugin with NTLARS revision history; shopfloor dashboard upcoming cards and state chips
This commit is contained in:
308
plugins/backups/services/ntlars.py
Normal file
308
plugins/backups/services/ntlars.py
Normal file
@@ -0,0 +1,308 @@
|
||||
"""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')
|
||||
Reference in New Issue
Block a user