108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
"""Parsers for extension root package pointers and CAS manifests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import re
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .payload import GUID_RE, parse_brace_text, payload_to_text, scalar
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExtensionZippedInfo:
|
|
marker_hex: str
|
|
root_cas_key: str
|
|
text_fragment: str
|
|
guids: list[str]
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManifestEntry:
|
|
object_id: str
|
|
cas_key: str
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def _extract_utf16le_fragment(data: bytes) -> str:
|
|
starts = [pos for pos in (data.find(b"{\x00"), data.find(b'"\x00#\x00"\x00')) if pos >= 0]
|
|
if not starts:
|
|
return ""
|
|
start = min(starts)
|
|
fragment = data[start:]
|
|
if len(fragment) % 2:
|
|
fragment = fragment[:-1]
|
|
return fragment.decode("utf-16-le", errors="ignore").strip("\x00")
|
|
|
|
|
|
def parse_extension_zipped_info(data: bytes) -> ExtensionZippedInfo:
|
|
text = _extract_utf16le_fragment(data)
|
|
return ExtensionZippedInfo(
|
|
marker_hex=data[:4].hex(),
|
|
root_cas_key=data[4:24].hex() if len(data) >= 24 else "",
|
|
text_fragment=text,
|
|
guids=sorted(set(match.lower() for match in GUID_RE.findall(text))),
|
|
)
|
|
|
|
|
|
def _base64_to_sha1(value: str) -> str | None:
|
|
if not re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", value):
|
|
return None
|
|
try:
|
|
data = base64.b64decode(value, validate=True)
|
|
except Exception:
|
|
return None
|
|
return data.hex() if len(data) == 20 else None
|
|
|
|
|
|
def parse_extension_manifest_bytes(data: bytes) -> dict[str, Any]:
|
|
decoded = payload_to_text(data)
|
|
text = decoded.get("text")
|
|
if text is None:
|
|
raise ValueError("cannot decode extension manifest")
|
|
parsed = parse_brace_text(text.lstrip("ï»¿п»ї"))
|
|
if not (isinstance(parsed, dict) and parsed.get("type") == "sequence"):
|
|
raise ValueError("expected extension root manifest sequence")
|
|
items = parsed.get("items") or []
|
|
if len(items) == 4 and scalar(items[0]) in {"", "п»ї"}:
|
|
items = items[1:]
|
|
if len(items) < 3:
|
|
raise ValueError("expected at least 3 sequence items")
|
|
|
|
payload_block = items[1]
|
|
manifest_block = items[2]
|
|
extension_guid = ""
|
|
if isinstance(payload_block, dict) and payload_block.get("type") == "list":
|
|
block_items = payload_block.get("items") or []
|
|
if len(block_items) > 1:
|
|
extension_guid = scalar(block_items[1]).lower()
|
|
|
|
manifest_items = manifest_block.get("items") if isinstance(manifest_block, dict) else []
|
|
declared_count = int(scalar(manifest_items[0]) or "0") if manifest_items else 0
|
|
entries: list[ManifestEntry] = []
|
|
for index in range(1, len(manifest_items or []), 2):
|
|
object_id = scalar(manifest_items[index])
|
|
encoded_key = scalar(manifest_items[index + 1]) if index + 1 < len(manifest_items) else ""
|
|
cas_key = _base64_to_sha1(encoded_key)
|
|
if object_id and cas_key:
|
|
entries.append(ManifestEntry(object_id=object_id.lower(), cas_key=cas_key))
|
|
|
|
return {
|
|
"compression": decoded["compression"],
|
|
"encoding": decoded["encoding"],
|
|
"extension_configuration_guid": extension_guid,
|
|
"declared_count": declared_count,
|
|
"entries": entries,
|
|
}
|
|
|
|
|
|
def parse_extension_manifest_file(path: Path) -> dict[str, Any]:
|
|
return parse_extension_manifest_bytes(path.read_bytes())
|