304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""Classify 1C Config/ConfigCAS payload parts without infobase-specific names."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import re
|
|
from typing import Any
|
|
|
|
from .payload import collect_strings, decode_payload_lossless, encode_text, parse_brace_text, root_signature, scalar
|
|
|
|
|
|
BASE64_RE = re.compile(r"[A-Za-z0-9+/]{40,}={0,2}")
|
|
BSL_MARKERS = ("&На", "Процедура ", "Функция ", "#Область", "#КонецОбласти")
|
|
HTML_MARKERS = ("<!DOCTYPE", "<html", "<HTML", "<body", "<BODY")
|
|
STREAM_HEADER_RE = re.compile(rb"\r\n([0-9a-f]{8}) ([0-9a-f]{8}) 7fffffff \r\n")
|
|
|
|
|
|
def sha1_hex(data: bytes) -> str:
|
|
return hashlib.sha1(data).hexdigest()
|
|
|
|
|
|
def normalized_text_sha1(text: str) -> str:
|
|
normalized = str(text or "").replace("\r\n", "\n").replace("\r", "\n")
|
|
return hashlib.sha1(normalized.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def decode_text(data: bytes) -> tuple[str | None, str | None]:
|
|
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]
|
|
marker_score = sum(500 for marker in (*BSL_MARKERS, *HTML_MARKERS) if marker in sample)
|
|
printable = sum(1 for char in sample if char.isprintable() or char in "\r\n\t")
|
|
score = printable + marker_score - sample.count("\x00") * 10
|
|
if score > best[2]:
|
|
best = (text, encoding, score)
|
|
return best[0], best[1]
|
|
|
|
|
|
def payload_markers(payload: bytes) -> list[str]:
|
|
markers = []
|
|
if payload.startswith(b"MOXCEL"):
|
|
markers.append("MOXCEL")
|
|
if payload.startswith(b"\xef\xbb\xbf") or b"\xef\xbb\xbf" in payload[:256]:
|
|
markers.append("utf8_bom")
|
|
if STREAM_HEADER_RE.search(payload):
|
|
markers.append("stream_headers")
|
|
return markers
|
|
|
|
|
|
def extract_stream_blocks(payload: bytes, *, include_text: bool = False, limit: int = 100) -> list[dict[str, Any]]:
|
|
blocks: list[dict[str, Any]] = []
|
|
for match in STREAM_HEADER_RE.finditer(payload):
|
|
declared_1 = int(match.group(1), 16)
|
|
declared_2 = int(match.group(2), 16)
|
|
start = match.end()
|
|
size = declared_2
|
|
if size <= 0 or start + size > len(payload):
|
|
continue
|
|
data = payload[start : start + size]
|
|
text, encoding = decode_text(data)
|
|
clean = (text or "").replace("\x00", "")
|
|
item: dict[str, Any] = {
|
|
"header_offset": match.start(),
|
|
"data_offset": start,
|
|
"declared_1": declared_1,
|
|
"declared_2": declared_2,
|
|
"bytes": len(data),
|
|
"sha1": sha1_hex(data),
|
|
"encoding": encoding,
|
|
"text_preview": clean[:500],
|
|
"has_bsl_marker": bool(text and any(marker in clean for marker in BSL_MARKERS)),
|
|
"has_html_marker": bool(text and any(marker in clean for marker in HTML_MARKERS)),
|
|
}
|
|
if include_text:
|
|
item["text"] = text
|
|
blocks.append(item)
|
|
if len(blocks) >= limit:
|
|
break
|
|
return blocks
|
|
|
|
|
|
def stream_blocks_with_data(payload: bytes, *, limit: int = 100) -> list[dict[str, Any]]:
|
|
blocks: list[dict[str, Any]] = []
|
|
for match in STREAM_HEADER_RE.finditer(payload):
|
|
declared_1 = int(match.group(1), 16)
|
|
declared_2 = int(match.group(2), 16)
|
|
start = match.end()
|
|
size = declared_2
|
|
if size <= 0 or start + size > len(payload):
|
|
continue
|
|
data = payload[start : start + size]
|
|
text, encoding = decode_text(data)
|
|
blocks.append(
|
|
{
|
|
"header_offset": match.start(),
|
|
"header_end": match.end(),
|
|
"data_offset": start,
|
|
"data_end": start + size,
|
|
"declared_1": declared_1,
|
|
"declared_2": declared_2,
|
|
"bytes": len(data),
|
|
"sha1": sha1_hex(data),
|
|
"encoding": encoding,
|
|
"text": text,
|
|
"data": data,
|
|
}
|
|
)
|
|
if len(blocks) >= limit:
|
|
break
|
|
return blocks
|
|
|
|
|
|
def stream_header(size: int) -> bytes:
|
|
if size < 0 or size > 0xFFFFFFFF:
|
|
raise ValueError("stream size is outside 8-hex header range")
|
|
encoded = f"{size:08x}"
|
|
return f"\r\n{encoded} {encoded} 7fffffff \r\n".encode("ascii")
|
|
|
|
|
|
def replace_stream_block(
|
|
payload: bytes,
|
|
stream_index: int,
|
|
*,
|
|
text: str | None = None,
|
|
data: bytes | None = None,
|
|
replace: dict[str, Any] | None = None,
|
|
routine: dict[str, Any] | None = None,
|
|
expected_contains: str | None = None,
|
|
expected_text_sha1: str | None = None,
|
|
) -> tuple[bytes, dict[str, Any]]:
|
|
blocks = stream_blocks_with_data(payload)
|
|
if stream_index < 0 or stream_index >= len(blocks):
|
|
raise IndexError(f"stream_index {stream_index} is outside {len(blocks)} stream blocks")
|
|
block = blocks[stream_index]
|
|
old_data = bytes(block["data"])
|
|
old_text = block.get("text")
|
|
encoding = block.get("encoding")
|
|
if expected_contains and (old_text is None or expected_contains not in old_text):
|
|
raise ValueError("expected_contains was not found in stream text")
|
|
old_text_sha1 = normalized_text_sha1(old_text or "") if old_text is not None else None
|
|
if expected_text_sha1 and (old_text_sha1 is None or expected_text_sha1.lower() != old_text_sha1):
|
|
raise ValueError("expected_text_sha1 does not match current stream text")
|
|
if replace is not None:
|
|
if old_text is None:
|
|
raise ValueError("stream text is not decodable")
|
|
old = str(replace.get("old") or "")
|
|
new = str(replace.get("new") or "")
|
|
if not old:
|
|
raise ValueError("replace.old is required")
|
|
count = int(replace.get("count") or 1)
|
|
if old not in old_text:
|
|
raise ValueError("replace.old was not found in stream text")
|
|
text = old_text.replace(old, new, count)
|
|
routine_edit = None
|
|
if routine is not None:
|
|
if old_text is None:
|
|
raise ValueError("stream text is not decodable")
|
|
from .bsl_validation import replace_routine_text
|
|
|
|
text, routine_edit = replace_routine_text(
|
|
old_text,
|
|
str(routine.get("text") or ""),
|
|
operation=str(routine.get("operation") or "replace"),
|
|
name=str(routine.get("name")) if routine.get("name") else None,
|
|
expected_old_sha1=str(routine.get("expected_old_sha1")) if routine.get("expected_old_sha1") else None,
|
|
expected_old_contains=str(routine.get("expected_old_contains")) if routine.get("expected_old_contains") else None,
|
|
)
|
|
if data is None:
|
|
if text is None:
|
|
raise ValueError("text, data, replace, or routine is required")
|
|
data = encode_text(text, encoding)
|
|
header = stream_header(len(data))
|
|
new_payload = payload[: block["header_offset"]] + header + data + payload[block["data_end"] :]
|
|
return new_payload, {
|
|
"stream_index": stream_index,
|
|
"encoding": encoding,
|
|
"old_sha1": sha1_hex(old_data),
|
|
"new_sha1": sha1_hex(data),
|
|
"old_text_sha1": old_text_sha1,
|
|
"new_text_sha1": normalized_text_sha1(text) if text is not None else None,
|
|
"old_bytes": len(old_data),
|
|
"new_bytes": len(data),
|
|
"old_text_preview": (old_text or "")[:500],
|
|
"new_text_preview": (text or "")[:500] if text is not None else None,
|
|
**({"routine": routine_edit} if routine_edit else {}),
|
|
}
|
|
|
|
|
|
def collect_base64_blocks(value: Any) -> list[str]:
|
|
blocks: list[str] = []
|
|
|
|
def walk(node: Any) -> None:
|
|
if isinstance(node, dict) and node.get("type") == "list":
|
|
items = node.get("items") or []
|
|
if items and scalar(items[0]) == "#base64":
|
|
chunks = [scalar(item) for item in items[1:] if BASE64_RE.fullmatch(scalar(item))]
|
|
if chunks:
|
|
blocks.append("".join(chunks))
|
|
for child in items:
|
|
walk(child)
|
|
|
|
walk(value)
|
|
return blocks
|
|
|
|
|
|
def decode_base64_blocks(blocks: list[str], *, include_text: bool = False, limit: int = 50) -> list[dict[str, Any]]:
|
|
result: list[dict[str, Any]] = []
|
|
for value in blocks[:limit]:
|
|
try:
|
|
data = base64.b64decode(value, validate=True)
|
|
except Exception:
|
|
continue
|
|
text, encoding = decode_text(data)
|
|
clean = (text or "").replace("\x00", "")
|
|
item: dict[str, Any] = {
|
|
"block_length": len(value),
|
|
"bytes": len(data),
|
|
"sha1": sha1_hex(data),
|
|
"encoding": encoding,
|
|
"text_preview": clean[:500],
|
|
"has_bsl_marker": bool(text and any(marker in clean for marker in BSL_MARKERS)),
|
|
"has_html_marker": bool(text and any(marker in clean for marker in HTML_MARKERS)),
|
|
}
|
|
if include_text:
|
|
item["text"] = text
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def classify_role(root_marker: str | None, markers: list[str], text: str | None, stream_blocks: list[dict[str, Any]], base64_blocks: list[dict[str, Any]]) -> str:
|
|
clean = (text or "").replace("\x00", "")
|
|
if root_marker == "1":
|
|
return "metadata_payload"
|
|
if root_marker == "4":
|
|
return "form_payload"
|
|
if root_marker == "5" or any(block.get("has_html_marker") for block in base64_blocks):
|
|
return "help_or_html_payload"
|
|
if root_marker == "8" or "MOXCEL" in markers:
|
|
return "template_payload"
|
|
if any(block.get("has_bsl_marker") for block in stream_blocks) or any(marker in clean for marker in BSL_MARKERS):
|
|
return "bsl_module_payload"
|
|
if "stream_headers" in markers:
|
|
return "stream_container"
|
|
if root_marker:
|
|
return "brace_payload"
|
|
return "binary_or_unknown_payload"
|
|
|
|
|
|
def classify_payload(data: bytes, *, include_text: bool = False, include_tree: bool = False) -> dict[str, Any]:
|
|
decoded = decode_payload_lossless(data)
|
|
payload = decoded.get("payload") if isinstance(decoded.get("payload"), (bytes, bytearray)) else b""
|
|
markers = payload_markers(bytes(payload))
|
|
stream_blocks = extract_stream_blocks(bytes(payload), include_text=include_text)
|
|
text = decoded.get("text")
|
|
tree = None
|
|
root = None
|
|
base64_decoded: list[dict[str, Any]] = []
|
|
strings_sample: list[str] = []
|
|
if text and "{" in text and "stream_headers" not in markers:
|
|
try:
|
|
tree = parse_brace_text(text)
|
|
root = root_signature(tree)
|
|
strings_sample = collect_strings(tree, limit=80)
|
|
base64_decoded = decode_base64_blocks(collect_base64_blocks(tree), include_text=include_text)
|
|
except Exception:
|
|
tree = None
|
|
root_marker = root.get("root_marker") if isinstance(root, dict) else None
|
|
result: dict[str, Any] = {
|
|
"status": "ok" if payload else "undecodable",
|
|
"compression": decoded.get("compression"),
|
|
"encoding": decoded.get("encoding"),
|
|
"raw_bytes": decoded.get("raw_bytes"),
|
|
"payload_bytes": decoded.get("payload_bytes"),
|
|
"sha1": sha1_hex(data),
|
|
"payload_sha1": sha1_hex(bytes(payload)),
|
|
"markers": markers,
|
|
"root": root,
|
|
"role": classify_role(root_marker, markers, text, stream_blocks, base64_decoded),
|
|
"strings_sample": strings_sample,
|
|
"stream_blocks": stream_blocks,
|
|
"base64_blocks": base64_decoded,
|
|
"counts": {
|
|
"stream_blocks": len(stream_blocks),
|
|
"base64_blocks": len(base64_decoded),
|
|
"strings_sample": len(strings_sample),
|
|
},
|
|
}
|
|
if include_text:
|
|
result["text"] = text
|
|
if include_tree:
|
|
result["tree"] = tree
|
|
return result
|