554 lines
21 KiB
Python
554 lines
21 KiB
Python
"""Low-level payload decoding and brace-tree parsing for 1C SQL metadata."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gzip
|
|
import copy
|
|
import re
|
|
import zlib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
GUID_RE = re.compile(
|
|
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
|
|
)
|
|
|
|
BraceNode = dict[str, Any]
|
|
|
|
|
|
def try_decompress(data: bytes) -> tuple[bytes, str]:
|
|
"""Try known 1C payload compression envelopes."""
|
|
|
|
attempts = (
|
|
("raw_deflate", lambda value: zlib.decompress(value, -15)),
|
|
("zlib", zlib.decompress),
|
|
("gzip", gzip.decompress),
|
|
)
|
|
for name, func in attempts:
|
|
try:
|
|
return func(data), name
|
|
except Exception:
|
|
pass
|
|
return data, "none"
|
|
|
|
|
|
def compress_payload(data: bytes, compression: str) -> bytes:
|
|
"""Compress payload bytes using a known 1C storage envelope."""
|
|
|
|
if compression == "raw_deflate":
|
|
compressor = zlib.compressobj(level=6, wbits=-15)
|
|
return compressor.compress(data) + compressor.flush()
|
|
if compression == "zlib":
|
|
return zlib.compress(data)
|
|
if compression == "gzip":
|
|
return gzip.compress(data)
|
|
if compression == "none":
|
|
return data
|
|
raise ValueError(f"unsupported compression: {compression}")
|
|
|
|
|
|
def try_decode(data: bytes) -> tuple[str | None, str | None]:
|
|
"""Decode payload text, preferring explicit BOM over heuristic scoring."""
|
|
|
|
if data.startswith(b"\xef\xbb\xbf"):
|
|
try:
|
|
return data.decode("utf-8-sig"), "utf-8-sig"
|
|
except UnicodeDecodeError:
|
|
pass
|
|
|
|
candidates = ("utf-8-sig", "utf-8", "utf-16-le", "utf-16-be", "cp1251")
|
|
best: tuple[str | None, str | None, int] = (None, None, -1)
|
|
for encoding in candidates:
|
|
try:
|
|
text = data.decode(encoding)
|
|
except UnicodeDecodeError:
|
|
continue
|
|
sample = text[:20000]
|
|
printable = sum(1 for char in sample if char.isprintable() or char in "\r\n\t")
|
|
cyrillic = sum(1 for char in sample if "\u0400" <= char <= "\u04ff")
|
|
cjk = sum(1 for char in sample if "\u4e00" <= char <= "\u9fff")
|
|
replacement = sample.count("\ufffd")
|
|
controls = sum(1 for char in sample if ord(char) < 32 and char not in "\r\n\t\x00")
|
|
score = printable + cyrillic * 4 - sample.count("\x00") * 10 - cjk * 12 - replacement * 20 - controls * 5
|
|
if score > best[2]:
|
|
best = (text, encoding, score)
|
|
return best[0], best[1]
|
|
|
|
|
|
def encode_text(text: str, encoding: str | None) -> bytes:
|
|
if not encoding:
|
|
raise ValueError("encoding is required")
|
|
return text.encode(encoding)
|
|
|
|
|
|
def decode_payload_lossless(data: bytes) -> dict[str, Any]:
|
|
"""Decode payload without stripping text bytes so it can be encoded back."""
|
|
|
|
payload, compression = try_decompress(data)
|
|
text, encoding = try_decode(payload)
|
|
return {
|
|
"payload": payload,
|
|
"compression": compression,
|
|
"text": text,
|
|
"encoding": encoding,
|
|
"raw_bytes": len(data),
|
|
"payload_bytes": len(payload),
|
|
}
|
|
|
|
|
|
def encode_payload_lossless(decoded: dict[str, Any], *, text: str | None = None, payload: bytes | None = None) -> bytes:
|
|
"""Encode a decoded payload using its original compression/encoding metadata."""
|
|
|
|
compression = str(decoded.get("compression") or "none")
|
|
if payload is None:
|
|
if text is None:
|
|
text = decoded.get("text")
|
|
payload = encode_text(str(text), decoded.get("encoding")) if text is not None else decoded.get("payload")
|
|
if not isinstance(payload, (bytes, bytearray)):
|
|
raise ValueError("payload bytes or text are required")
|
|
return compress_payload(bytes(payload), compression)
|
|
|
|
|
|
def payload_to_text(data: bytes) -> dict[str, Any]:
|
|
"""Decode compressed SQL BinaryData to text and generic metadata."""
|
|
|
|
payload, compression = try_decompress(data)
|
|
text, encoding = try_decode(payload)
|
|
return {
|
|
"payload": payload,
|
|
"compression": compression,
|
|
"text": text.replace("\x00", "").replace("\ufeff", "") if text is not None else None,
|
|
"encoding": encoding,
|
|
"raw_bytes": len(data),
|
|
"payload_bytes": len(payload),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Token:
|
|
kind: str
|
|
value: str
|
|
pos: int
|
|
end: int
|
|
|
|
|
|
class Lexer:
|
|
def __init__(self, text: str) -> None:
|
|
self.text = text
|
|
self.pos = 0
|
|
|
|
def tokens(self) -> list[Token]:
|
|
result: list[Token] = []
|
|
while self.pos < len(self.text):
|
|
char = self.text[self.pos]
|
|
if char.isspace():
|
|
self.pos += 1
|
|
continue
|
|
if char in "{},:":
|
|
result.append(Token(char, char, self.pos, self.pos + 1))
|
|
self.pos += 1
|
|
continue
|
|
if char == '"':
|
|
result.append(self._string())
|
|
continue
|
|
result.append(self._atom())
|
|
result.append(Token("EOF", "", self.pos, self.pos))
|
|
return result
|
|
|
|
def _string(self) -> Token:
|
|
start = self.pos
|
|
self.pos += 1
|
|
chars: list[str] = []
|
|
while self.pos < len(self.text):
|
|
char = self.text[self.pos]
|
|
self.pos += 1
|
|
if char == '"':
|
|
if self.pos < len(self.text) and self.text[self.pos] == '"':
|
|
chars.append('"')
|
|
self.pos += 1
|
|
continue
|
|
break
|
|
chars.append(char)
|
|
return Token("string", "".join(chars), start, self.pos)
|
|
|
|
def _atom(self) -> Token:
|
|
start = self.pos
|
|
while self.pos < len(self.text):
|
|
char = self.text[self.pos]
|
|
if char.isspace() or char in "{},:":
|
|
break
|
|
self.pos += 1
|
|
return Token("atom", self.text[start : self.pos], start, self.pos)
|
|
|
|
|
|
class Parser:
|
|
def __init__(self, tokens: list[Token]) -> None:
|
|
self.tokens = tokens
|
|
self.index = 0
|
|
|
|
def parse(self) -> Any:
|
|
values = []
|
|
while not self._peek("EOF"):
|
|
if self._peek(","):
|
|
self.index += 1
|
|
continue
|
|
values.append(self._value())
|
|
if len(values) == 1:
|
|
return values[0]
|
|
return {"type": "sequence", "items": values}
|
|
|
|
def _value(self) -> Any:
|
|
if self._peek("{"):
|
|
return self._list()
|
|
token = self._next()
|
|
if token.kind == "string":
|
|
return {"type": "string", "value": token.value, "pos": token.pos, "end": token.end}
|
|
if token.kind == "atom":
|
|
return {"type": "atom", "value": token.value, "pos": token.pos, "end": token.end}
|
|
return {"type": "token", "kind": token.kind, "value": token.value, "pos": token.pos, "end": token.end}
|
|
|
|
def _list(self) -> Any:
|
|
start = self._next()
|
|
items = []
|
|
while not self._peek("EOF") and not self._peek("}"):
|
|
if self._peek(","):
|
|
self.index += 1
|
|
continue
|
|
items.append(self._value())
|
|
end = items[-1].get("end", items[-1].get("pos", start.end)) if items and isinstance(items[-1], dict) else start.end
|
|
if self._peek("}"):
|
|
end = self.tokens[self.index].end
|
|
self.index += 1
|
|
return {"type": "list", "pos": start.pos, "end": end, "items": items}
|
|
|
|
def _peek(self, kind: str) -> bool:
|
|
return self.tokens[self.index].kind == kind
|
|
|
|
def _next(self) -> Token:
|
|
token = self.tokens[self.index]
|
|
self.index += 1
|
|
return token
|
|
|
|
|
|
def parse_brace_text(text: str) -> Any:
|
|
"""Parse brace text into a generic tree without semantic interpretation."""
|
|
|
|
clean = text.replace("\x00", "").replace("\ufeff", "")
|
|
first_brace = clean.find("{")
|
|
if first_brace > 0:
|
|
clean = clean[first_brace:]
|
|
return Parser(Lexer(clean).tokens()).parse()
|
|
|
|
|
|
def quote_string(value: str) -> str:
|
|
return '"' + value.replace('"', '""') + '"'
|
|
|
|
|
|
def patch_brace_text_path(text: str, path: str, value: Any, *, node_type: str = "auto") -> tuple[str, dict[str, Any]]:
|
|
"""Patch one scalar token in-place without canonicalizing the whole brace tree."""
|
|
|
|
if "\x00" in text:
|
|
raise ValueError("byte-preserving patch does not support NUL-stripped payload text")
|
|
offset = text.find("{")
|
|
if offset < 0:
|
|
raise ValueError("brace text payload is required")
|
|
source = text[offset:]
|
|
tree = Parser(Lexer(source).tokens()).parse()
|
|
old = get_tree_path(tree, path)
|
|
if not isinstance(old, dict) or old.get("type") not in {"atom", "string"}:
|
|
raise ValueError("byte-preserving patch supports scalar atom/string nodes only")
|
|
replacement_type = node_type
|
|
if replacement_type == "auto":
|
|
replacement_type = str(old.get("type") or "string")
|
|
replacement = scalar_node(value, replacement_type)
|
|
if replacement.get("type") == "string":
|
|
rendered = quote_string(str(replacement.get("value") or ""))
|
|
elif replacement.get("type") == "atom":
|
|
rendered = str(replacement.get("value") or "")
|
|
else:
|
|
raise ValueError("byte-preserving patch replacement must be atom or string")
|
|
start = offset + int(old.get("pos"))
|
|
end = offset + int(old.get("end"))
|
|
if start < 0 or end < start or end > len(text):
|
|
raise ValueError("invalid scalar token span")
|
|
patched = text[:start] + rendered + text[end:]
|
|
return patched, {
|
|
"path": path,
|
|
"old": scalar(old),
|
|
"new": str(replacement.get("value") or ""),
|
|
"old_node_type": old.get("type"),
|
|
"new_node_type": replacement.get("type"),
|
|
"span": [start, end],
|
|
}
|
|
|
|
|
|
def swap_brace_text_paths(text: str, path_a: str, path_b: str) -> tuple[str, dict[str, Any]]:
|
|
"""Swap two parsed brace-tree node text spans without canonicalizing the tree."""
|
|
|
|
if "\x00" in text:
|
|
raise ValueError("byte-preserving swap does not support NUL-stripped payload text")
|
|
offset = text.find("{")
|
|
if offset < 0:
|
|
raise ValueError("brace text payload is required")
|
|
if path_a == path_b:
|
|
raise ValueError("swap paths must be different")
|
|
parts_a = tree_path_parts(path_a)
|
|
parts_b = tree_path_parts(path_b)
|
|
if parts_a == parts_b[: len(parts_a)] or parts_b == parts_a[: len(parts_b)]:
|
|
raise ValueError("swap paths must not be ancestor/descendant")
|
|
source = text[offset:]
|
|
tree = Parser(Lexer(source).tokens()).parse()
|
|
node_a = get_tree_path(tree, path_a)
|
|
node_b = get_tree_path(tree, path_b)
|
|
if not isinstance(node_a, dict) or not isinstance(node_b, dict):
|
|
raise ValueError("swap paths must resolve to parsed nodes")
|
|
if "pos" not in node_a or "end" not in node_a or "pos" not in node_b or "end" not in node_b:
|
|
raise ValueError("swap nodes must have text spans")
|
|
span_a = [offset + int(node_a["pos"]), offset + int(node_a["end"])]
|
|
span_b = [offset + int(node_b["pos"]), offset + int(node_b["end"])]
|
|
if span_a[0] >= span_a[1] or span_b[0] >= span_b[1]:
|
|
raise ValueError("swap node span is empty")
|
|
if not (span_a[1] <= span_b[0] or span_b[1] <= span_a[0]):
|
|
raise ValueError("swap node spans overlap")
|
|
first_span, second_span = (span_a, span_b) if span_a[0] < span_b[0] else (span_b, span_a)
|
|
first_text = text[first_span[0] : first_span[1]]
|
|
second_text = text[second_span[0] : second_span[1]]
|
|
swapped = text[: first_span[0]] + second_text + text[first_span[1] : second_span[0]] + first_text + text[second_span[1] :]
|
|
return swapped, {
|
|
"path_a": path_a,
|
|
"path_b": path_b,
|
|
"span_a": span_a,
|
|
"span_b": span_b,
|
|
"node_a_type": node_a.get("type"),
|
|
"node_b_type": node_b.get("type"),
|
|
}
|
|
|
|
|
|
def append_brace_text_child(text: str, parent_path: str, child_node: Any) -> tuple[str, dict[str, Any]]:
|
|
"""Append one child node to a parsed brace-tree container without reformatting the rest."""
|
|
|
|
if "\x00" in text:
|
|
raise ValueError("byte-preserving append does not support NUL-stripped payload text")
|
|
offset = text.find("{")
|
|
if offset < 0:
|
|
raise ValueError("brace text payload is required")
|
|
source = text[offset:]
|
|
tree = Parser(Lexer(source).tokens()).parse()
|
|
parent = get_tree_path(tree, parent_path)
|
|
if not isinstance(parent, dict) or parent.get("type") not in {"list", "sequence"}:
|
|
raise ValueError("append parent must be a list or sequence node")
|
|
if "end" not in parent:
|
|
raise ValueError("append parent span is unavailable")
|
|
items = parent.get("items") if isinstance(parent.get("items"), list) else []
|
|
insertion = offset + int(parent["end"]) - 1
|
|
if insertion < 0 or insertion > len(text) or text[insertion] != "}":
|
|
raise ValueError("append insertion point is not a closing brace")
|
|
rendered_child = serialize_brace_tree(child_node)
|
|
inserted = ("," if items else "") + rendered_child
|
|
count_updated = False
|
|
count_before = None
|
|
count_after = None
|
|
count_span = None
|
|
original_insertion = insertion
|
|
prefix = text[:original_insertion]
|
|
if len(items) >= 2:
|
|
count_node = items[1]
|
|
count_text = scalar(count_node)
|
|
try:
|
|
declared_count = int(count_text) if count_text is not None else None
|
|
except ValueError:
|
|
declared_count = None
|
|
actual_record_count = len(items) - 2
|
|
if declared_count is not None and declared_count == actual_record_count and isinstance(count_node, dict) and "pos" in count_node and "end" in count_node:
|
|
count_start = offset + int(count_node["pos"])
|
|
count_end = offset + int(count_node["end"])
|
|
if 0 <= count_start < count_end <= len(prefix):
|
|
count_before = declared_count
|
|
count_after = declared_count + 1
|
|
count_span = [count_start, count_end]
|
|
prefix = prefix[:count_start] + str(count_after) + prefix[count_end:]
|
|
count_updated = True
|
|
insertion = len(prefix)
|
|
patched = prefix + inserted + text[original_insertion:]
|
|
return patched, {
|
|
"parent_path": parent_path,
|
|
"inserted_index": len(items),
|
|
"span": [insertion, insertion],
|
|
"inserted_bytes": len(inserted.encode("utf-8")),
|
|
"count_updated": count_updated,
|
|
**({"count_before": count_before, "count_after": count_after, "count_span": count_span} if count_updated else {}),
|
|
}
|
|
|
|
|
|
def serialize_brace_tree(node: Any) -> str:
|
|
"""Serialize parsed brace tree to canonical 1C brace text."""
|
|
|
|
if isinstance(node, dict):
|
|
node_type = node.get("type")
|
|
if node_type == "list":
|
|
return "{" + ",".join(serialize_brace_tree(item) for item in (node.get("items") or [])) + "}"
|
|
if node_type == "sequence":
|
|
return ",".join(serialize_brace_tree(item) for item in (node.get("items") or []))
|
|
if node_type == "string":
|
|
return quote_string(str(node.get("value") or ""))
|
|
if node_type == "atom":
|
|
return str(node.get("value") or "")
|
|
if node_type == "token":
|
|
return str(node.get("value") or "")
|
|
if isinstance(node, str):
|
|
return quote_string(node)
|
|
if node is None:
|
|
return ""
|
|
return str(node)
|
|
|
|
|
|
def tree_path_parts(path: str) -> list[int]:
|
|
if not str(path or "").strip():
|
|
raise ValueError("path is required")
|
|
parts: list[int] = []
|
|
for part in str(path).split("."):
|
|
if not part.isdigit():
|
|
raise ValueError(f"path segment is not a non-negative integer: {part}")
|
|
parts.append(int(part))
|
|
return parts
|
|
|
|
|
|
def get_tree_path(tree: Any, path: str) -> Any:
|
|
node = tree
|
|
for index in tree_path_parts(path):
|
|
if not isinstance(node, dict) or node.get("type") not in {"list", "sequence"}:
|
|
raise ValueError(f"path enters a non-container node at segment {index}")
|
|
items = node.get("items") or []
|
|
if index >= len(items):
|
|
raise IndexError(f"path segment {index} is outside node with {len(items)} items")
|
|
node = items[index]
|
|
return node
|
|
|
|
|
|
def inspect_tree_node(node: Any, *, depth: int = 0, max_depth: int = 2, max_children: int = 8) -> dict[str, Any]:
|
|
if not isinstance(node, dict):
|
|
return {"type": type(node).__name__, "repr": repr(node)[:200]}
|
|
node_type = str(node.get("type") or "")
|
|
result: dict[str, Any] = {"type": node_type}
|
|
for key in ("value", "pos", "end", "kind"):
|
|
if key in node:
|
|
result[key] = node.get(key)
|
|
items = node.get("items") if isinstance(node.get("items"), list) else None
|
|
if items is not None:
|
|
result["items"] = len(items)
|
|
if depth < max_depth:
|
|
result["children"] = [
|
|
inspect_tree_node(child, depth=depth + 1, max_depth=max_depth, max_children=max_children)
|
|
for child in items[:max_children]
|
|
]
|
|
return result
|
|
|
|
|
|
def inspect_brace_text_path(text: str, path: str, *, max_depth: int = 2, max_children: int = 8) -> dict[str, Any]:
|
|
if "\x00" in text:
|
|
raise ValueError("byte-preserving probe does not support NUL-stripped payload text")
|
|
offset = text.find("{")
|
|
if offset < 0:
|
|
raise ValueError("brace text payload is required")
|
|
source = text[offset:]
|
|
tree = Parser(Lexer(source).tokens()).parse()
|
|
node = get_tree_path(tree, path)
|
|
result = {"path": path, "node": inspect_tree_node(node, max_depth=max_depth, max_children=max_children)}
|
|
if isinstance(node, dict) and "pos" in node:
|
|
result["span"] = [offset + int(node.get("pos")), offset + int(node.get("end", node.get("pos")))]
|
|
return result
|
|
|
|
|
|
def scalar_node(value: Any, node_type: str = "auto") -> dict[str, Any]:
|
|
if isinstance(value, dict) and value.get("type") in {"atom", "string", "list", "sequence", "token"}:
|
|
return value
|
|
if node_type not in {"auto", "atom", "string"}:
|
|
raise ValueError("node_type must be auto, atom, or string")
|
|
if node_type == "atom":
|
|
return {"type": "atom", "value": str(value)}
|
|
if node_type == "string":
|
|
return {"type": "string", "value": str(value)}
|
|
if isinstance(value, bool):
|
|
return {"type": "atom", "value": "true" if value else "false"}
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return {"type": "atom", "value": str(value)}
|
|
return {"type": "string", "value": str(value)}
|
|
|
|
|
|
def set_tree_path(tree: Any, path: str, value: Any, *, node_type: str = "auto") -> Any:
|
|
"""Return a deep-copied tree with one node replaced by path."""
|
|
|
|
parts = tree_path_parts(path)
|
|
result = copy.deepcopy(tree)
|
|
parent = result
|
|
for index in parts[:-1]:
|
|
if not isinstance(parent, dict) or parent.get("type") not in {"list", "sequence"}:
|
|
raise ValueError(f"path enters a non-container node at segment {index}")
|
|
items = parent.get("items") or []
|
|
if index >= len(items):
|
|
raise IndexError(f"path segment {index} is outside node with {len(items)} items")
|
|
parent = items[index]
|
|
if not isinstance(parent, dict) or parent.get("type") not in {"list", "sequence"}:
|
|
raise ValueError("path parent is not a container")
|
|
items = parent.get("items") or []
|
|
last = parts[-1]
|
|
if last >= len(items):
|
|
raise IndexError(f"path segment {last} is outside node with {len(items)} items")
|
|
old = items[last]
|
|
replacement_type = node_type
|
|
if replacement_type == "auto" and isinstance(old, dict) and old.get("type") in {"atom", "string"}:
|
|
replacement_type = str(old.get("type"))
|
|
items[last] = scalar_node(value, replacement_type)
|
|
return result
|
|
|
|
|
|
def encode_brace_tree(tree: Any, decoded: dict[str, Any]) -> bytes:
|
|
"""Encode a modified brace tree with canonical formatting."""
|
|
|
|
return encode_payload_lossless(decoded, text=serialize_brace_tree(tree))
|
|
|
|
|
|
def parse_payload_file(path: Path) -> dict[str, Any]:
|
|
decoded = payload_to_text(path.read_bytes())
|
|
text = decoded.get("text")
|
|
decoded["tree"] = parse_brace_text(text) if text and "{" in text else None
|
|
return decoded
|
|
|
|
|
|
def scalar(node: Any) -> str:
|
|
if isinstance(node, dict) and node.get("type") in {"atom", "string"}:
|
|
return str(node.get("value") or "")
|
|
return ""
|
|
|
|
|
|
def collect_strings(value: Any, limit: int = 200) -> list[str]:
|
|
result: list[str] = []
|
|
|
|
def walk(node: Any) -> None:
|
|
if len(result) >= limit:
|
|
return
|
|
if isinstance(node, dict) and node.get("type") == "string":
|
|
text = str(node.get("value") or "")
|
|
if text:
|
|
result.append(text)
|
|
return
|
|
if isinstance(node, dict):
|
|
for child in node.get("items") or []:
|
|
walk(child)
|
|
|
|
walk(value)
|
|
return result
|
|
|
|
|
|
def root_signature(tree: Any) -> dict[str, Any]:
|
|
if not (isinstance(tree, dict) and tree.get("type") == "list"):
|
|
return {"root_type": tree.get("type") if isinstance(tree, dict) else None}
|
|
items = tree.get("items") or []
|
|
return {
|
|
"root_type": "list",
|
|
"root_len": len(items),
|
|
"root_marker": scalar(items[0]) if items else "",
|
|
}
|