#!/usr/bin/env python3
"""
sesame2 all-in-one tool
=======================
Consolidated single-file version of the ZoneDirector sesame2 toolkit
(previously split across sesame2_keygen.py, patch_sesame2_duration.py and
the one-shot orchestrator).
Modes
-----
one-shot : patch_sesame2.py <sesame2-bin> [out] --serial <N>
re-key the binary with a fresh random EC key pair, patch
out the validity-window check, and print a passphrase for
the serial on stdout (all diagnostics go to stderr).
The new private key is saved to <out>.key.pem -- keep it
to mint passphrases for other serials.
check-only : patch_sesame2.py --check-only <sesame2-bin> [out]
only patch out the validity-window check (no re-key).
keygen : patch_sesame2.py --keygen-only <serial> [--privkey P] [...]
mint a passphrase with an existing private key (e.g. the
<out>.key.pem saved by a one-shot run).
Every binary location is done by byte signature with cross-validation, so
the patchers work across ZoneDirector firmware builds that share the same
sesame2 source/toolchain family. If a binary is not recognizable, the tool
aborts WITHOUT modifying anything.
Crypto model (verified step-by-step against the binary under qemu+gdb):
AES-128 key K = 19 dd 5c 4c 93 31 04 9d 34 12 00 06 70 79 45 51
mix const IV = be ca f0 43 b0 a2 3d 84 31 94 ba 97 2c 66 de bd
PAT = 4a d7 6f 32 63 41 42 07
digest = MD5( salt_byte || serial_string ) (serial = 12-digit)
s = AES_K(0x00*16)
s = AES_K( s ^ IV )
s = AES_K( s ^ (PAT || digest[0:8]) )
s[0:8] ^= digest[8:16] ; s[8] += 0x80
L = AES_K(0) (CMAC-style subkeys)
K1 = (L<<1) ^ (0x87 if MSB(L)) ; K2 = (K1<<1) ^ (0x87 if MSB(K1))
counter0 = AES_K( K2 ^ s )
keystream_i = AES_K(counter0 + i) (big-endian 128-bit increment)
Key string: custom base64 ('.'=0, '0'-'9'=1..10, 'A'-'Z'=12..37,
'a'-'z'=38..63, value 11 unused). The 3 control-field chars
[hex(salt)][version] sit at position (sum of last 4 chars)%10 and are
REMOVED by the binary (memmove in copy_key_field) before base64-decoding,
so they are free-form. blob_len = 4 + len(sig) + kv must be %3==0.
Plaintext: [4-byte header][ECDSA-SHA1 DER signature over the header];
header[0]=(info_len<<4)|((duration>>8)&1), info_len==4, header[1]=duration
(9-bit), header[2..3]=start days since 2014-08-01 (16-bit). The signed
"info" is the 4 header bytes themselves.
Curve: secp128r1 (OID 1.3.132.0.28) on the ZD 10.5.1 binary.
"""
import argparse
import base64
import datetime
import hashlib
import os
import struct
import subprocess
import sys
import tempfile
# --------------------------------------------------------------------------
# crypto constants + AES
# --------------------------------------------------------------------------
K = bytes.fromhex("19dd5c4c9331049d3412000670794551")
IV = bytes.fromhex("becaf043b0a23d843194ba972c66debd")
PAT = bytes.fromhex("4ad76f3263414207")
EPOCH2014 = datetime.date(2014, 8, 1)
CURVE_OIDS = { # OIDs seen in ZoneDirector sesame2 builds
"1.3.132.0.28": "secp128r1",
"1.2.840.10045.3.1.7": "prime256v1",
"1.3.132.0.10": "secp256k1",
"1.3.132.0.33": "secp224r1",
"1.3.132.0.34": "secp384r1",
"1.3.132.0.35": "secp521r1",
}
class AESCtx:
"""Reusable AES-128-ECB encryptor (the binary's aes_block_round).
Imported lazily so check-only mode works without python-cryptography."""
def __init__(self):
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
self._enc = Cipher(algorithms.AES(K), modes.ECB()).encryptor()
def block(self, data):
assert len(data) == 16
return self._enc.update(data)
# --------------------------------------------------------------------------
# custom base64 (alphabet: '.'=0, '0'-'9'=1..10, 'A'-'Z'=12..37,
# 'a'-'z'=38..63; value 11 has no character)
# --------------------------------------------------------------------------
def _b64_char(value):
if value == 0:
return "."
if 1 <= value <= 10:
return chr(0x30 + value - 1)
if 12 <= value <= 37:
return chr(ord("A") + value - 12)
if 38 <= value <= 63:
return chr(ord("a") + value - 38)
return None # value 11: unencodable
def encode_chunk(b3):
"""3 bytes -> 4 chars. Raises ValueError when a sextet == 11."""
v = int.from_bytes(b3, "big")
sextets = [(v >> 18) & 63, (v >> 12) & 63, (v >> 6) & 63, v & 63]
out = []
for x in sextets:
c = _b64_char(x)
if c is None:
raise ValueError("sextet 11")
out.append(c)
return "".join(out)
def custom_b64encode(data):
"""Encode; data length must be a multiple of 3 (blob_len % 3 == 0)."""
assert len(data) % 3 == 0
return "".join(encode_chunk(data[i:i + 3]) for i in range(0, len(data), 3))
def decode_key(s):
"""Custom-base64 decode (mirrors the binary's decoder state machine)."""
vals = []
for ch in s:
if ch == ".":
vals.append(0)
elif "0" <= ch <= "9":
vals.append(ord(ch) - 0x2F)
elif "A" <= ch <= "Z":
vals.append(ord(ch) - 0x35)
elif "a" <= ch <= "z":
vals.append(ord(ch) - 0x3B)
else:
raise ValueError("bad char %r" % ch)
out = bytearray()
acc = 0
state = 0
for v in vals:
if state == 0:
acc = v * 4
state = 6
elif state == 6:
out.append((v >> 4) | acc)
acc = (v & 0xF) << 4
state = 4
elif state == 4:
out.append((v >> 2) | acc)
acc = (v & 3) << 6
state = 2
else:
out.append(acc | v)
state = 0
return bytes(out)
# --------------------------------------------------------------------------
# counter derivation + CTR
# --------------------------------------------------------------------------
def derive_counter(serial, salt_byte, aes):
digest = hashlib.md5(bytes([salt_byte]) + serial.encode("ascii")).digest()
s = aes.block(b"\x00" * 16)
s = bytes(a ^ b for a, b in zip(s, IV))
s = aes.block(s)
s = bytes(a ^ b for a, b in zip(s, PAT + digest[:8]))
s = bytearray(aes.block(s))
for i in range(8):
s[i] ^= digest[8 + i]
s[8] = (s[8] + 0x80) & 0xFF
L = aes.block(b"\x00" * 16)
def subkey(x):
msb = x[0] & 0x80
v = (int.from_bytes(x, "big") << 1) & ((1 << 128) - 1)
if msb:
v ^= 0x87
return v.to_bytes(16, "big")
K2 = subkey(subkey(L))
state = bytes(a ^ b for a, b in zip(K2, bytes(s)))
return aes.block(state)
def ctr_keystream(counter0, nblocks, aes):
ks = bytearray()
c = int.from_bytes(counter0, "big")
for _ in range(nblocks):
ks += aes.block(c.to_bytes(16, "big"))
c = (c + 1) & ((1 << 128) - 1)
return bytes(ks)
# --------------------------------------------------------------------------
# passphrase generation
# --------------------------------------------------------------------------
def make_signature(privkey_pem, header, workdir):
"""ECDSA-SHA1 DER signature over `header` via host openssl (has
secp128r1; python-cryptography does not)."""
sig_path = os.path.join(workdir, "sig.bin")
info_path = os.path.join(workdir, "info.bin")
with open(info_path, "wb") as f:
f.write(header)
subprocess.run(
["openssl", "dgst", "-sha1", "-sign", privkey_pem,
"-out", sig_path, info_path],
check=True, capture_output=True,
)
with open(sig_path, "rb") as f:
return f.read()
def _check_digit(s):
"""(sum of the LAST 4 chars) % 10 -- verified against the disassembly."""
return sum(ord(ch) for ch in s[-4:]) % 10
def build_header(start_days, duration_days):
info_len = 4
return bytes([
(info_len << 4) | ((duration_days >> 8) & 1),
duration_days & 0xFF,
(start_days >> 8) & 0xFF,
start_days & 0xFF,
])
def generate_key(serial, privkey_pem, start_days, duration_days, workdir,
salt=0x43, verbose=True):
"""Build the passphrase string. Returns (key, header, sig, kv)."""
if not (0 <= start_days <= 0xFFFF and 1 <= duration_days <= 0x1FF):
raise ValueError("start 0..65535, duration 1..511")
aes = AESCtx()
for attempt in range(32):
header = build_header(start_days, duration_days)
sig = make_signature(privkey_pem, header, workdir)
S = len(sig)
kv = (0 - (4 + S)) % 3 # so that (4 + S + kv) % 3 == 0
plaintext = header + sig
counter0 = derive_counter(serial, salt, aes)
ks = ctr_keystream(counter0, (len(plaintext) + 15) // 16, aes)
ct = bytes(a ^ b for a, b in zip(plaintext, ks))
blob = ct + bytes(kv) # kv junk bytes (any values)
try:
s0 = custom_b64encode(blob)
except ValueError:
continue # unencodable sextet (value 11)
c = _check_digit(s0)
control = "%02x%s" % (salt, "024"[kv])
full = s0[:c] + control + s0[c:]
if len(full) < 50:
continue
if _check_digit(full) == c:
if verbose:
print(f"[+] key len={len(full)} sig_len={S} kv={kv} "
f"c={c} salt={salt:02x} start={start_days} "
f"dur={duration_days}", file=sys.stderr)
return full, header, sig, kv
raise RuntimeError("failed to build a valid key string")
def verify_key(serial, s, pubkey_pem=None, verbose=True):
"""Mirror the binary exactly (remove control chars, decode, decrypt)
and optionally check the ECDSA signature with openssl."""
c = _check_digit(s)
ctrl = s[c:c + 3]
salt_byte = int(ctrl[0:2], 16)
kv = (int(ctrl[2], 16) & 6) >> 1
if verbose:
print(f"[*] check_digit={c} control='{ctrl}' salt={salt_byte:02x} kv={kv}")
if kv == 3:
raise ValueError("kv==3 rejected by binary")
payload = s[:c] + s[c + 3:] # the binary memmoves these out
blob = decode_key(payload)
if verbose:
print(f"[*] decoded {len(payload)} chars -> {len(blob)} bytes")
blob = blob[:len(blob) - kv]
aes = AESCtx()
counter0 = derive_counter(serial, salt_byte, aes)
nblocks = (len(blob) + 15) // 16
ks = ctr_keystream(counter0, nblocks, aes)
pt = bytes(a ^ b for a, b in zip(blob, ks))
info_len = pt[0] >> 4
duration = ((pt[0] & 1) << 8) | pt[1]
start = (pt[2] << 8) | pt[3]
sig = pt[info_len:]
print(f"[*] plaintext ({len(pt)} B): {pt.hex()}")
print(f"[*] header: info_len={info_len} start_days={start} "
f"duration_days={duration}")
print(f"[*] window: {EPOCH2014 + datetime.timedelta(days=start)} .. "
f"{EPOCH2014 + datetime.timedelta(days=start + duration)}")
if pubkey_pem:
with tempfile.TemporaryDirectory() as td:
ip = os.path.join(td, "info.bin")
sp = os.path.join(td, "sig.bin")
with open(ip, "wb") as f:
f.write(pt[:info_len])
with open(sp, "wb") as f:
f.write(sig)
r = subprocess.run(
["openssl", "dgst", "-sha1", "-verify", pubkey_pem,
"-signature", sp, ip], capture_output=True)
print(f"[*] openssl verify: {r.stdout.decode().strip()}")
return pt
# --------------------------------------------------------------------------
# ELF32 helpers (enough to get .text / .rodata)
# --------------------------------------------------------------------------
def load_sections(data):
"""Return {name: (vaddr, file_offset, size)} for ELF32 sections."""
if data[:4] != b"\x7fELF":
raise ValueError("not an ELF file")
if data[4] != 1: # ELFCLASS32
raise ValueError("not a 32-bit ELF")
e_shoff = struct.unpack_from("<I", data, 0x20)[0]
e_shentsize = struct.unpack_from("<H", data, 0x2E)[0]
e_shnum = struct.unpack_from("<H", data, 0x30)[0]
e_shstrndx = struct.unpack_from("<H", data, 0x32)[0]
def shent(i):
off = e_shoff + i * e_shentsize
name, typ, flags, addr, ofs, size = struct.unpack_from("<IIIIII", data, off)
return name, typ, addr, ofs, size
_, _, _, shstr_off, shstr_size = shent(e_shstrndx)
shstr = data[shstr_off:shstr_off + shstr_size]
def name_at(n):
end = shstr.index(b"\x00", n)
return shstr[n:end].decode()
out = {}
for i in range(e_shnum):
nm, typ, addr, ofs, size = shent(i)
if typ == 1 and size: # SHT_PROGBITS
out[name_at(nm)] = (addr, ofs, size)
return out
def find_all(hay, needle):
out = []
i = hay.find(needle)
while i >= 0:
out.append(i)
i = hay.find(needle, i + 1)
return out
# --------------------------------------------------------------------------
# window-check patch: byte signatures (portable across ZD builds)
# --------------------------------------------------------------------------
FP_A1 = bytes.fromhex("d9eedae9dfe09e") # fldz;fucompp;fnstsw;sahf
FP_A2 = bytes.fromhex("d9eed9c9dae9dfe0") # fldz;fxch;fucompp;fnstsw
MOV77 = bytes.fromhex("bb4d000000") # mov $0x4d,%ebx (exit 77)
TEST_JLE = b"\x85\xc0\x7e" # test %eax,%eax ; jle disp8
CALL = 0xE8
ENTRY_ALREADY = bytes.fromhex("b801000000c3") # our own patch marker
rodata_bytes = b"" # set by the CLI before analyze()
already_found = False # set by analyze()
def find_function_entry(text, interior_off):
"""Entry of the function containing `interior_off`: the nearest preceding
`push ebp; mov ebp,esp` prologue within 256 bytes -- OR our own
already-applied marker `b8 01 00 00 00 c3` (mov $1,%eax;ret), which is
nearer than any prologue once the check is patched.
Returns (entry_off, already_patched)."""
base = max(0, interior_off - 256)
window = text[base:interior_off]
i_already = window.rfind(ENTRY_ALREADY)
i_pro = window.rfind(b"\x55\x89\xe5")
if i_already >= 0 and i_already > i_pro:
return base + i_already, True
if i_pro >= 0:
return base + i_pro, False
return None, False
def analyze(data, sections):
"""Locate and validate the check; return a list of patch plans.
Sets module globals rodata_bytes/already_found."""
global already_found
text_vma, text_off, text_size = sections[".text"]
text = data[text_off:text_off + text_size]
rodata = sections.get(".rodata")
ro_vma = ro_off = None
if rodata:
ro_vma, ro_off, _ = rodata
a1 = find_all(text, FP_A1)
a2 = find_all(text, FP_A2)
print(f"[.] FP idiom A1 (fldz;fucompp;fnstsw;sahf): {len(a1)} match(es)")
print(f"[.] FP idiom A2 (fldz;fxch;fucompp;fnstsw): {len(a2)} match(es)")
func_entries = set()
already_entries = set()
for off in a1 + a2:
entry, already = find_function_entry(text, off)
if entry is not None:
func_entries.add(entry)
if already:
already_entries.add(entry)
# A must have both idioms inside one function
valid_func = None
for e in func_entries:
if any(a1) and any(a2):
near = lambda o: e <= o < e + 0x200
if any(near(o) for o in a1) and any(near(o) for o in a2):
valid_func = e
break
plans = []
if valid_func is not None:
entry_vma = text_vma + valid_func
already = valid_func in already_entries
already_found = already
print(f"[+] is_within_validity entry @ {entry_vma:#x} "
f"({'already patched' if already else 'patching to mov $1,%eax; ret'})")
if not already:
plans.append(("func_entry", entry_vma, ENTRY_ALREADY,
"is_within_validity always returns 1"))
else:
print("[!] FP-idiom function not reliably located; trying string xref fallback")
# B: call-site idiom, cross-validated against the function we found
b_plans = []
for off in find_all(text, MOV77):
vma = text_vma + off
call_tgt = None
for back in range(1, 65):
if text[off - back] == CALL:
disp = struct.unpack_from("<i", text, off - back + 1)[0]
call_tgt = text_vma + (off - back) + 5 + disp
break
if valid_func is not None and (text_vma + valid_func) != call_tgt:
tgt_s = hex(call_tgt) if call_tgt else "None"
print(f"[~] mov $0x4d,%ebx @ {vma:#x}: preceding call -> {tgt_s} "
f"(not our function); skipped")
continue
if text[off + 5:off + 8] != TEST_JLE:
print(f"[~] mov $0x4d,%ebx @ {vma:#x}: no test;jle after; skipped")
continue
tgt_s = hex(call_tgt) if call_tgt else "None"
print(f"[+] exit-77 idiom @ {vma:#x} (call->{tgt_s}); "
f"patching xor ebx,ebx + NOP test;jle")
b_plans.append(("call_site", vma, MOV77, "call site patched"))
b_plans.append(("nop_jle", vma + 5, b"\x90\x90\x90\x90", "test;jle NOPed"))
plans += b_plans
# C: string-xref fallback (only if we still have nothing)
if not b_plans and rodata is not None:
for needle in (b"Expiration: %F %T", b"%F %T"):
i = rodata_bytes.find(needle)
if i >= 0:
str_vma = ro_vma + i
pat = struct.pack("<I", str_vma)
for x in find_all(text, pat):
lo = max(0, x - 128)
j = text.rfind(MOV77, lo, x)
if j >= 0 and text[j + 5:j + 8] == TEST_JLE:
vma = text_vma + j
print(f"[+] (fallback) exit-77 idiom @ {vma:#x} near "
f"Expiration string xref {text_vma + x:#x}")
plans.append(("call_site", vma, MOV77, "call site patched"))
plans.append(("nop_jle", vma + 5, b"\x90\x90\x90\x90",
"test;jle NOPed"))
break
if plans:
break
return plans
def apply_patches(data, sections, plans, out_path):
text_vma, text_off, _ = sections[".text"]
for kind, vma, payload, desc in plans:
off = text_off + (vma - text_vma)
old = bytes(data[off:off + len(payload)])
data[off:off + len(payload)] = payload
print(f" patched {kind} @ {vma:#x}: {old.hex()} -> {payload.hex()} ({desc})")
open(out_path, "wb").write(data)
print(f"[+] wrote {out_path} ({len(data)} bytes, {len(plans)} patch(es))")
# --------------------------------------------------------------------------
# re-key helpers: find the embedded PEM, parse its curve, generate a new pair
# --------------------------------------------------------------------------
def der_oid(der, pos):
"""Parse a DER OBJECT IDENTIFIER at pos -> (oid_string, next_pos)."""
assert der[pos] == 0x06, "expected OBJECT IDENTIFIER"
ln = der[pos + 1]
content = der[pos + 2:pos + 2 + ln]
first = content[0]
parts = [first // 40, first % 40]
i = 1
while i < len(content):
v = 0
while True:
b = content[i]
i += 1
v = (v << 7) | (b & 0x7F)
if not (b & 0x80):
break
parts.append(v)
return ".".join(str(p) for p in parts), pos + 2 + ln
def spki_curve_oid(der):
"""Curve OID from a SubjectPublicKeyInfo DER blob."""
assert der[0] == 0x30, "not a SEQUENCE"
pos = 2
assert der[pos] == 0x30, "missing AlgorithmIdentifier"
pos += 2
_, pos = der_oid(der, pos) # id-ecPublicKey (1.2.840.10045.2.1)
oid, _ = der_oid(der, pos) # the named curve OID
return oid
def make_public_pem(der):
"""Normalize a DER SPKI to PEM with 64-char base64 lines."""
body = base64.b64encode(der).decode()
lines = [body[i:i + 64] for i in range(0, len(body), 64)]
return ("-----BEGIN PUBLIC KEY-----\n" + "\n".join(lines) +
"\n-----END PUBLIC KEY-----\n")
def find_embedded_pem(rodata):
i = rodata.find(b"-----BEGIN PUBLIC KEY-----")
if i < 0:
return None
end = rodata.find(b"\x00", i)
if end < 0:
return None
return rodata[i:end], i
def generate_keypair(curve_name, privkey_out, workdir):
"""Fresh random EC key pair via host openssl; returns the public DER."""
subprocess.run(["openssl", "ecparam", "-name", curve_name, "-genkey",
"-noout", "-out", privkey_out],
check=True, capture_output=True)
pub_path = os.path.join(workdir, "newpub.pem")
subprocess.run(["openssl", "ec", "-in", privkey_out, "-pubout",
"-out", pub_path],
check=True, capture_output=True)
with open(pub_path, "rb") as f:
body = b"".join(f.read().split(b"\n")[1:-2])
return base64.b64decode(body)
# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------
def mode_one_shot(args):
"""Re-key + remove the window check + print a passphrase for --serial."""
if len(args.serial) not in (10, 11, 12) or not args.serial.isdigit():
print(f"[!] --serial should be a digit string (12-digit); got "
f"{args.serial!r}", file=sys.stderr)
sys.exit(2)
out = args.out or args.src
privkey_out = args.privkey_out or (out + ".key.pem")
data = bytearray(open(args.src, "rb").read())
sections = load_sections(data)
if ".text" not in sections or ".rodata" not in sections:
print("[!] missing .text/.rodata; not a usable sesame2 binary",
file=sys.stderr)
sys.exit(1)
ro_vma, ro_off, ro_size = sections[".rodata"]
rodata = data[ro_off:ro_off + ro_size]
# 1) find the embedded public key ---------------------------------------
pem, pem_ro = find_embedded_pem(rodata)
if pem is None:
print("[!] no '-----BEGIN PUBLIC KEY-----' found in .rodata",
file=sys.stderr)
sys.exit(1)
body = b"".join(pem.split(b"\n")[1:-2])
old_der = base64.b64decode(body)
oid = spki_curve_oid(old_der)
curve = CURVE_OIDS.get(oid)
print(f"[.] embedded public key @ rodata+{pem_ro:#x} "
f"({len(pem)} bytes), curve OID {oid} -> {curve or 'UNKNOWN'}",
file=sys.stderr)
if curve is None:
print(f"[!] unsupported curve OID {oid}; add it to CURVE_OIDS",
file=sys.stderr)
sys.exit(1)
# 2) window-check analysis FIRST (read-only); abort before mutating -----
saved_stdout = sys.stdout
sys.stdout = sys.stderr # keep stdout clean for the passphrase
try:
global rodata_bytes, already_found
rodata_bytes = rodata
already_found = False
plans = analyze(data, sections)
if not plans and not already_found:
print("[!] duration-check signature not found; no changes made",
file=sys.stderr)
sys.exit(1)
# 3) generate a fresh key pair on the same curve --------------------
new_der = generate_keypair(curve, privkey_out, args.workdir)
new_pem = make_public_pem(new_der).encode()
if len(new_pem) != len(pem):
print(f"[!] new PEM is {len(new_pem)} bytes but embedded is "
f"{len(pem)}; in-place patch not possible for curve {curve}",
file=sys.stderr)
sys.exit(1)
print(f"[+] generated fresh {curve} key pair -> {privkey_out}",
file=sys.stderr)
# 4) apply the patches ----------------------------------------------
data[ro_off + pem_ro:ro_off + pem_ro + len(pem)] = new_pem
print(f"[+] patched embedded public key @ rodata+{pem_ro:#x}",
file=sys.stderr)
if plans:
apply_patches(data, sections, plans, out)
else:
open(out, "wb").write(data)
print("[+] duration check already patched -- no changes made",
file=sys.stderr)
finally:
sys.stdout = saved_stdout
# 5) mint + print the passphrase ---------------------------------------
key, header, sig, kv = generate_key(
args.serial, privkey_out, args.start, args.duration, args.workdir,
salt=args.salt)
print(key) # the passphrase -- the only stdout line
print(f"# serial={args.serial} start={args.start} duration={args.duration} "
f"kv={kv} sig_len={len(sig)} salt={args.salt:02x} "
f"header={header.hex()} curve={curve}", file=sys.stderr)
print(f"# patched binary: {out} | private key: {privkey_out} "
f"(keep it to mint more passphrases)", file=sys.stderr)
def mode_check_only(args):
"""Only patch out the validity-window check."""
out = args.out or args.src
data = bytearray(open(args.src, "rb").read())
sections = load_sections(data)
if ".text" not in sections:
print("[!] no .text section; not a usable sesame2 binary", file=sys.stderr)
sys.exit(1)
rodata = sections.get(".rodata")
global rodata_bytes, already_found
rodata_bytes = data[rodata[1]:rodata[1] + rodata[2]] if rodata else b""
already_found = False
plans = analyze(data, sections)
if not plans and not already_found:
print("[!] no recognizable duration-check signature found -- "
"binary may be a very different build; no changes made",
file=sys.stderr)
sys.exit(1)
if plans:
apply_patches(data, sections, plans, out)
else:
print("[+] duration check already patched -- no changes made",
file=sys.stderr)
def mode_keygen(args):
"""Mint a passphrase with an existing private key (--privkey)."""
if len(args.serial) not in (10, 11, 12) or not args.serial.isdigit():
print(f"[!] serial should be a digit string (12-digit); got "
f"{args.serial!r}", file=sys.stderr)
sys.exit(2)
key, header, sig, kv = generate_key(
args.serial, args.privkey, args.start, args.duration, args.workdir,
salt=args.salt)
print(key) # the passphrase -- the only stdout line
print(f"# serial={args.serial} start={args.start} duration={args.duration} "
f"kv={kv} sig_len={len(sig)} salt={args.salt:02x} "
f"header={header.hex()}", file=sys.stderr)
if args.verify:
pub_path = os.path.join(args.workdir, "verify_pub.pem")
subprocess.run(["openssl", "ec", "-in", args.privkey, "-pubout",
"-out", pub_path], check=True, capture_output=True)
verify_key(args.serial, key, pubkey_pem=pub_path)
def main():
ap = argparse.ArgumentParser(
description="ZoneDirector sesame2 tool: re-key, remove the validity-"
"window check, and/or mint passphrases.",
usage="%(prog)s <sesame2-bin> [out] --serial <N>\n"
" %(prog)s --check-only <sesame2-bin> [out]\n"
" %(prog)s --keygen-only <serial> [options]")
ap.add_argument("src", nargs="?", help="sesame2 binary (or serial with "
"--keygen-only)")
ap.add_argument("out", nargs="?", default=None,
help="output path (default: patch in place)")
ap.add_argument("--check-only", action="store_true",
help="only patch out the validity-window check")
ap.add_argument("--keygen-only", action="store_true",
help="only mint a passphrase (src is the serial)")
ap.add_argument("--serial", metavar="N",
help="12-digit serial; one-shot mode also generates a "
"passphrase for it")
ap.add_argument("--privkey", default="sesame_privkey.pem",
help="private key PEM for signing (keygen/one-shot sign "
"with the NEW key saved by one-shot)")
ap.add_argument("--privkey-out", default=None,
help="where to save the NEW private key (one-shot; "
"default: <output>.key.pem)")
ap.add_argument("--salt", type=lambda x: int(x, 0), default=0x43,
help="salt byte for MD5(salt||serial) (default: 0x43)")
ap.add_argument("--start", type=int, default=None,
help="validity start in days since 2014-08-01 "
"(default: today-45; cosmetic once the check is gone)")
ap.add_argument("--duration", type=int, default=90,
help="validity length in days, 1..511 (default: 90)")
ap.add_argument("--workdir", default=None,
help="temp dir for openssl (default: system temp)")
ap.add_argument("--verify", action="store_true",
help="keygen-only: self-check the passphrase with openssl")
args = ap.parse_args()
if args.workdir is None:
args.workdir = tempfile.gettempdir()
if args.start is None:
args.start = (datetime.date.today() - EPOCH2014).days - 45
modes = [args.check_only, args.keygen_only, bool(args.serial)]
if sum(modes) != 1:
ap.error("exactly one mode required: --check-only, --keygen-only, "
"or --serial (one-shot)")
if args.check_only:
if not args.src:
ap.error("--check-only needs a sesame2 binary path")
mode_check_only(args)
elif args.keygen_only:
if not args.src:
ap.error("--keygen-only needs a serial number")
args.serial = args.src
mode_keygen(args)
else:
if not args.src:
ap.error("one-shot mode needs a sesame2 binary path")
mode_one_shot(args)
if __name__ == "__main__":
main()