109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Conservative parser for top-level Config metadata object identity."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .payload import GUID_RE, parse_payload_file, root_signature, scalar
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MetadataObjectIdentity:
|
|
guid: str
|
|
name: str
|
|
synonyms: dict[str, str]
|
|
evidence_path: str
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def _is_guid(value: str) -> bool:
|
|
return bool(GUID_RE.fullmatch(value))
|
|
|
|
|
|
def _identity_guid(node: Any) -> str:
|
|
if not (isinstance(node, dict) and node.get("type") == "list"):
|
|
return ""
|
|
items = node.get("items") or []
|
|
if len(items) != 3:
|
|
return ""
|
|
if scalar(items[0]) != "1" or scalar(items[1]) != "0":
|
|
return ""
|
|
guid = scalar(items[2]).lower()
|
|
return guid if _is_guid(guid) else ""
|
|
|
|
|
|
def _synonyms(node: Any) -> dict[str, str]:
|
|
if not (isinstance(node, dict) and node.get("type") == "list"):
|
|
return {}
|
|
items = node.get("items") or []
|
|
if not items:
|
|
return {}
|
|
try:
|
|
declared_count = int(scalar(items[0]))
|
|
except ValueError:
|
|
return {}
|
|
if declared_count < 0 or len(items) < 1 + declared_count * 2:
|
|
return {}
|
|
result = {}
|
|
index = 1
|
|
end = 1 + declared_count * 2
|
|
while index + 1 < end:
|
|
language = scalar(items[index])
|
|
value = scalar(items[index + 1])
|
|
if language and value:
|
|
result[language] = value
|
|
index += 2
|
|
return result
|
|
|
|
|
|
def find_identity(tree: Any) -> MetadataObjectIdentity | None:
|
|
"""Find the observed object identity block in a generic brace tree."""
|
|
|
|
def walk(node: Any, path: list[int]) -> MetadataObjectIdentity | None:
|
|
if isinstance(node, dict) and node.get("type") == "list":
|
|
items = node.get("items") or []
|
|
for index in range(0, max(len(items) - 2, 0)):
|
|
guid = _identity_guid(items[index])
|
|
name = scalar(items[index + 1])
|
|
synonym_node = items[index + 2]
|
|
synonym_items = synonym_node.get("items") if isinstance(synonym_node, dict) and synonym_node.get("type") == "list" else None
|
|
try:
|
|
synonym_count = int(scalar(synonym_items[0])) if synonym_items else -1
|
|
except ValueError:
|
|
synonym_count = -1
|
|
synonyms = _synonyms(synonym_node)
|
|
if guid and name and synonym_count >= 0 and len(synonym_items or []) >= 1 + synonym_count * 2:
|
|
return MetadataObjectIdentity(
|
|
guid=guid,
|
|
name=name,
|
|
synonyms=synonyms,
|
|
evidence_path=".".join(str(part) for part in [*path, index]),
|
|
)
|
|
for child_index, child in enumerate(items):
|
|
found = walk(child, [*path, child_index])
|
|
if found:
|
|
return found
|
|
return None
|
|
|
|
return walk(tree, [])
|
|
|
|
|
|
def parse_config_object_file(path: Path) -> dict[str, Any]:
|
|
payload = parse_payload_file(path)
|
|
tree = payload.get("tree")
|
|
identity = find_identity(tree)
|
|
return {
|
|
"source_path": str(path),
|
|
"compression": payload.get("compression"),
|
|
"encoding": payload.get("encoding"),
|
|
"raw_bytes": payload.get("raw_bytes"),
|
|
"payload_bytes": payload.get("payload_bytes"),
|
|
"root": root_signature(tree),
|
|
"identity": identity.to_dict() if identity else None,
|
|
"tree": tree,
|
|
}
|