init
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
"""Patch campaign / channel id embedded in type-0x01 secondary dylibs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
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}$")
|
||||
|
||||
|
||||
def gen_channel_id() -> str:
|
||||
"""32 lowercase hex chars (same width as campaign slot)."""
|
||||
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"(got {value!r})"
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def find_channel_offsets(data: bytes, *, expect_old: bytes | None = None) -> list[int]:
|
||||
"""Return offsets of the 32-byte channel value after CHANNEL_ANCHOR."""
|
||||
offsets: list[int] = []
|
||||
start = 0
|
||||
while True:
|
||||
index = data.find(CHANNEL_ANCHOR, start)
|
||||
if index < 0:
|
||||
break
|
||||
value_at = index + len(CHANNEL_ANCHOR)
|
||||
value = data[value_at : value_at + CHANNEL_LEN]
|
||||
if (
|
||||
len(value) == CHANNEL_LEN
|
||||
and _CHANNEL_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:
|
||||
offsets.append(value_at)
|
||||
start = index + 1
|
||||
return offsets
|
||||
|
||||
|
||||
def patch_channel_in_dylib(
|
||||
data: bytes,
|
||||
new_channel: str,
|
||||
*,
|
||||
old_channel: str | None = None,
|
||||
expect_hits: int = 1,
|
||||
label: str = "dylib",
|
||||
) -> bytes:
|
||||
"""In-place replace 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
|
||||
if old_b is not None:
|
||||
validate_channel_id(old_channel, name="old channel")
|
||||
|
||||
offsets = find_channel_offsets(data, expect_old=old_b)
|
||||
if len(offsets) != expect_hits:
|
||||
raise SystemExit(
|
||||
f"{label}: channel anchor hits={len(offsets)} (want {expect_hits}). "
|
||||
"Secondary payload layout may have changed; re-adapt _channel_patch."
|
||||
)
|
||||
|
||||
buf = bytearray(data)
|
||||
for offset in offsets:
|
||||
buf[offset : offset + CHANNEL_LEN] = new_b
|
||||
return bytes(buf)
|
||||
Reference in New Issue
Block a user