coruna-lab/channel-builder/tools/_common.py

"""Shared paths and seed helpers for channel-builder tooling.

Layout (coruna-lab/channel-builder):
  source/
  tools/
  Served artifacts default to Laravel public/:
    public/web/<channel-id>/
    public/sync/
  Builder state defaults to storage/app/channel-builder/:
    lab_seeds.json, out/
"""

from __future__ import annotations

import hashlib
import sys
from pathlib import Path

TOOLS_ROOT = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS_ROOT.parent
# Historical alias: LAB_ROOT == builder root (source/ + tools/)
LAB_ROOT = BUILDER_ROOT
FRONTEND_ROOT = BUILDER_ROOT  # legacy name used by some scripts
PROJECT_ROOT = BUILDER_ROOT.parent  # coruna-lab
SOURCE_ROOT = BUILDER_ROOT / "source"
ARTIFACTS_ROOT = PROJECT_ROOT / "public"
STATE_ROOT = PROJECT_ROOT / "storage" / "app" / "channel-builder"
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.
# Shared layout: root/sync + root/web/<channel>/
_TREE_ROOT = SOURCE_ROOT
_SHARED_LAYOUT = False
CAMPAIGN_DIR = _TREE_ROOT / "web"
SYNC_DIR = _TREE_ROOT / "sync"
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web"

# Vendored plaintext inputs (all under source/).
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 shared_layout() -> bool:
    return _SHARED_LAYOUT


def campaign_id() -> str:
    return CAMPAIGN_HASH


def set_campaign_id(channel: str) -> str:
    """Set the channel embedded in payloads; refresh CAMPAIGN_DIR for current layout."""
    global CAMPAIGN_HASH, CAMPAIGN_DIR
    from _channel_patch import validate_channel_id

    CAMPAIGN_HASH = validate_channel_id(channel, name="channel id")
    if _SHARED_LAYOUT:
        CAMPAIGN_DIR = _TREE_ROOT / "web" / CAMPAIGN_HASH
    else:
        CAMPAIGN_DIR = _TREE_ROOT / "web"
    return CAMPAIGN_HASH


def set_tree_root(
    root: Path,
    *,
    channel: str | None = None,
    shared_layout: bool = False,
) -> Path:
    """Point CAMPAIGN_DIR / SYNC_DIR at artifact root.

    shared_layout=True → root/web/<channel>/ + root/sync/
    shared_layout=False → root/web/ + root/sync/ (legacy isolated channel root)
    """
    global _TREE_ROOT, _SHARED_LAYOUT, CAMPAIGN_DIR, SYNC_DIR
    root = root.resolve()
    _TREE_ROOT = root
    _SHARED_LAYOUT = shared_layout
    SYNC_DIR = root / "sync"
    if channel is not None:
        set_campaign_id(channel)
    elif shared_layout:
        CAMPAIGN_DIR = root / "web" / CAMPAIGN_HASH
    else:
        CAMPAIGN_DIR = root / "web"
    return root


def ensure_tree_layout(
    root: Path,
    *,
    channel: str | None = None,
    require_campaign: bool = True,
    require_sync: bool = True,
) -> None:
    _ = channel  # channel already applied via set_tree_root
    if root.resolve() != _TREE_ROOT:
        raise SystemExit(f"ensure_tree_layout root mismatch: {root} != {_TREE_ROOT}")
    missing = []
    if require_campaign and not CAMPAIGN_DIR.is_dir():
        missing.append("web/" + (f"{CAMPAIGN_HASH}/" if _SHARED_LAYOUT else ""))
    if require_sync and not SYNC_DIR.is_dir():
        missing.append("sync/")
    if missing:
        raise SystemExit(
            f"missing working tree under {root} (need {', '.join(missing)}).\n"
            f"Build with:\n"
            f"  python3 tools/new_project.py --artifact-root {root} --channel-id <id>"
        )


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()