Ruckus Wireless as an Unifi alternative?

Notice: Page may contain affiliate links for which we may earn a small commission through services like Amazon Affiliates or Skimlinks.

int0x2e

Active Member
Dec 9, 2015
160
125
43
47
I have an R750 on unleashed.
Has anyone got the internal IoT (Zigbee/BT) antenna functionality working?
I'm told it's available if you buy an (enterprise) subscription - which is ridiculous for something we've already paid for!
I think the IoT capabilities are only available on the controller managed variant, not on Unleashed.
I did check out what they offered a very long time ago and didn't think it was that interesting personally. I just got a PoE powered zigbee2mqtt hub instead. It's obviously not nearly as sleekly integrated though...
 
  • Like
Reactions: abq and iceman_jkh

Rand__

Well-Known Member
Mar 6, 2014
6,723
1,827
113
I even have the add on USB Stick to enable this older models but never got it to work. I also only ever found the Enterpriose IOT management suite that does anything with IoT, so I considered this a dead end. And that's despite running a Zone Director and later Virtual SmartZone, so managed FW.
 
  • Like
Reactions: abq and iceman_jkh

dig

New Member
May 13, 2021
20
20
3
This is an experiemental virtual ZoneDirector 1200 running in a QEMU VM inside a Docker. It's not a polished product release, just a hack project and requires a bit of persistance to get working. With that caveat in mind...

I have been running it for several days alongside my existing Unleashed network. It has successfully adopted and upgraded a physical R600, but I do not have a physical ZoneDirector for comparison, and other AP models remain untested.

code and some brief instructions are here : https://github.com/dbro/zd1200

The project uses the decrypted Ruckus 10.5.1.0.282 software image with compatibility patches. No Ruckus binaries or firmware are included in the repository; users must obtain the software themselves. You should be familiar with docker and haveat least one spare AP that you can connect to it for testing.

I made this because I didn't have a hardware ZD1200 and wanted to get some new capabilties out of my R600 APs, primarily ethernet port VLAN settings. I also want to be able to add some R650 APs in the future, but there is no single version of Unleashed that supports R600 and R650 devices, but ZD v10.5.1 does support both of them. One drawback is that the ZD controller is a new single point of failure in the network, compared to Unleashed multi-master failover mode.

I plan to run it separately from my Unleashed network to evaluate its reliability and behavior and UX. It will be one of about 20 containers on a docker host with a 12th gen i3 and 16GB RAM, and the zd1200 container uses about 2GB (correction : ) 600MB of RAM and 3% of one CPU core. I plugged in a separate USB adapter to use as a dedicated ethernet port for the ZD1200, separate from the regular network connection used by the rest of the system and the other docker containers.

Feel free to check it out and it would be great to hear if/how it works for other people. It's MIT licensed so you can do what you want with it.

Btw, @ms264556 I am still interested in this alternative you mentioned previously if you're still working on it. I imagine it would be even better than this virtual ZD approach.
I'm working on a SmartZone compatible controller. I'll post here when I need more testers.
 
Last edited:

ms264556

Well-Known Member
Sep 13, 2021
522
504
93
New Zealand
ms264556.net
This is an experiemental virtual ZoneDirector 1200 running in a QEMU VM inside a Docker.
Wow. This is a huge amount of fantastic work! Can't wait to try it out.

Your readme says it requires a decrypted firmware. Do feel free to paste my decryption snippet into your script, or point people at an online decrypter.

And Ruckus provide an rdb script to set the ZD serial (and MACs). Will this work for your image?
 

ms264556

Well-Known Member
Sep 13, 2021
522
504
93
New Zealand
ms264556.net
The project uses the decrypted Ruckus 10.5.1.0.282 software image with compatibility patches.
I notice your scripts have some 10.2 compatibility points. If you tested against older versions then 10.3.1 and 10.1.2 are interesting releases to support too:
 

dig

New Member
May 13, 2021
20
20
3
Thanks @ms264556 for those suggestions, I added links to decrypt and set the serial/mac/etc in the github repo's README .

I notice your scripts have some 10.2 compatibility points. If you tested against older versions then 10.3.1 and 10.1.2 are interesting releases to support too:
You are right! I originally got it working with ZD software version 10.2.1.0.232, and then used that as a starting point to get this version working. the main difference IIRC was 10.2 used ftp for transferring images to the APs and that had some quirks. Let me know if you want me to send you the code for 10.2, but I don't intend to add support for other versions myself because I don't own those AP models. I previously owned a R700 (my first standalone AP) but sold it when I decided to have more than one AP and wanted to manage them all from a single console (Unleashed).
 
  • Love
Reactions: ms264556

unmesh

Active Member
Apr 17, 2017
280
74
28
67
This is an experiemental virtual ZoneDirector 1200 running in a QEMU VM inside a Docker. It's not a polished product release, just a hack project and requires a bit of persistance to get working. With that caveat in mind...

I have been running it for several days alongside my existing Unleashed network. It has successfully adopted and upgraded a physical R600, but I do not have a physical ZoneDirector for comparison, and other AP models remain untested.

,,,
I'm running Unleashed on several R600s and would like to replace only one of them with a WiFi6 AP. I also have a ProxMox server with underutilised CPU and DRAM.

This might be the way to do it so I will watch it with interest.

@dig, how long does it take for the ZD instance to boot? And since the ZD and R600 seem to be on an isolated L2 network, how are you connecting that to a router and the Internet? (I need to read up more on docker and network mapping)
 

dig

New Member
May 13, 2021
20
20
3
@dig, how long does it take for the ZD instance to boot? And since the ZD and R600 seem to be on an isolated L2 network, how are you connecting that to a router and the Internet? (I need to read up more on docker and network mapping)
Stopping the zd1200 docker container is instantaneous. Booting it takes about 35 seconds.

My home network has a few VLANs and the networking equipment uses one of them as a management VLAN. The docker host has a lot of other services that are all connected to a different VLAN, so I want to isolate this zd1200 container to talk only on the management VLAN via its own USB ethernet adapter. This USB adapter does not get its own IP address, it just passes traffic through it to the docker container and my DHCP server assigns an IP address to the zd1200 running on the VM inside the docker container using the MAC address set in the zd1200 firmware.

There is a managed switch that handles VLAN tagging and a firewall that blocks traffic between the management VLAN and the WAN/internet. I don't see any reason to allow the internal network equipment (the APs, ZD, and managed switch) to access the internet; they get their NTP time data from a server in my LAN and I don't want them to automatically check for updates or send any logs or data offsite.

The ZD only needs to connect to the management VLAN, and the managed switch handles VLAN tagging for that and does not need to send any other VLAN traffic to it.

The R600 AP needs a trunk port with tagged VLANs that include the management VLAN and all the other VLANs that want to be available to wireless clients.
 
Last edited:

dig

New Member
May 13, 2021
20
20
3
@ms264556 I'd like to have a convenient way to get a root shell in the ZD running in the VM. I think that should be possible by adding a new command to /bin/sys_wrapper.sh and later invoking it from the cli. Could you advise a good way to do that?
 

ms264556

Well-Known Member
Sep 13, 2021
522
504
93
New Zealand
ms264556.net
@ms264556 I'd like to have a convenient way to get a root shell in the ZD running in the VM. I think that should be possible by adding a new command to /bin/sys_wrapper.sh and later invoking it from the cli. Could you advise a good way to do that?
In my ZD root shell patches, I do this:-

Bash:
mount -o remount,rw /

cd /etc/persistent-scripts

cat <<EOF >.root.sh
#!/bin/sh
#RUCKUS#
/bin/stty echo
/bin/sh
EOF
chmod +x .root.sh
rm -f /writable/etc/scripts/.root.sh
ln -s -f /etc/persistent-scripts/.root.sh /writable/etc/scripts/.root.sh

mount -o remount,ro /
But if you don't want to muck about with your rootfs then you can just dump the script straight into /writable/etc/scripts/.root.sh and make it executable. Make sure you include those 2 magic comment lines at the top of your script, if you change it.

You get to the shell with:-

Code:
ruckus> enable
ruckus# debug
You have all rights in this mode.
ruckus(debug)# script
ruckus(script)# exec .root.sh
Ruckus Wireless ZoneDirector -- Command Line Interface
ruckus$
The nice thing to do (but lots more effort) would be to replace the sesame2 private key, so you can use the !v54! command to securely get a root shell. The key is plaintext inside the sesame executable, so this would be a safe thing to patch. If you get lots of people using your docker recipe, then maybe look at this.

One other thing which would be nice, is to generate an ECDSA SSH key so you can SSH in without reducing your client's security settings.

My ECDSA patch does this:-

Bash:
killall dropbear > /dev/null 2>&1
rm -f /etc/airespider/dropbear/dropbear_host_rsa_key > /dev/null 2>&1
dropbearkey -t ecdsa -f /etc/airespider/dropbear/dropbear_host_rsa_key > /dev/null 2>&1
/etc/init.d/dropbear > /dev/null 2>&1
 
Last edited:
  • Like
Reactions: abq and dig

dig

New Member
May 13, 2021
20
20
3
Thanks again @ms264556 ! Having root shell makes patching things like these so much easier. Without it, I needed to stop the zd1200 container, spin up a helper docker container and install and run guestfish to get to the zd1200's files inside the VM and read/edit them. The script you provided is working for me. Now to use it, for example:

Regarding sesame2 private key replacement, what is the value of the key string to search for and replace? Maybe sed from the root shell could do it.

Also instead of replacing the existing dropbear rsa key, I created a new ecdsa key file alongside the existing rsa key file and edited /etc/init.d/dropbear to add it as an accepted key. My goal was to simplify ssh into the zd1200 by supporting the current client defaults, not to get rid of rsa to improve the minimum security of ssh sessions.
 
Last edited:

unmesh

Active Member
Apr 17, 2017
280
74
28
67
Stopping the zd1200 docker container is instantaneous. Booting it takes about 35 seconds.

My home network has a few VLANs and the networking equipment uses one of them as a management VLAN. The docker host has a lot of other services that are all connected to a different VLAN, so I want to isolate this zd1200 container to talk only on the management VLAN via its own USB ethernet adapter. This USB adapter does not get its own IP address, it just passes traffic through it to the docker container and my DHCP server assigns an IP address to the zd1200 running on the VM inside the docker container using the MAC address set in the zd1200 firmware.

There is a managed switch that handles VLAN tagging and a firewall that blocks traffic between the management VLAN and the WAN/internet. I don't see any reason to allow the internal network equipment (the APs, ZD, and managed switch) to access the internet; they get their NTP time data from a server in my LAN and I don't want them to automatically check for updates or send any logs or data offsite.

The ZD only needs to connect to the management VLAN, and the managed switch handles VLAN tagging for that and does not need to send any other VLAN traffic to it.

The R600 AP needs a trunk port with tagged VLANs that include the management VLAN and all the other VLANs that want to be available to wireless clients.
I use VLANs too and my router/firewall is pfSense. My switches are unmanaged so don't themselves need Internet access.

Looks like you are passing one VLAN to the USB adapter after first creating an access port for it on your managed switch. The R600 gets the entire trunk.

Ruckus APs (at least those running Unleashed) do a heartbeat check by pinging "well known" servers on the Internet; if they can't reach any, they turn off wireless connectivity. How do you take care of this if the management VLAN does not have Internet access? I wonder if they try all VLANs.
 
Last edited:
  • Like
Reactions: dig

ms264556

Well-Known Member
Sep 13, 2021
522
504
93
New Zealand
ms264556.net
Regarding sesame2 private key replacement, what is the value of the key string to search for and replace? Maybe sed from the root shell could do it.
Thinking about it, this is not so useful for you.
When I was developing against the ZD, I built a statically linked dropbear (& sftp-server) and ran on port 2222 (via a /writable/data/scripts/start/
script) with with my public key in its authorized_keys. Not many friendly SSH/SFTP clients are happy with the hacky authentication and shell escape commands otherwise.
Probably a coding agent can quickly whip this up for you these days. Otherwise I can check my old ZD dev tools git history later in the week.

Also instead of replacing the existing dropbear rsa key, I created a new ecdsa key file alongside the existing rsa key file and edited /etc/init.d/dropbear to add it as an accepted key. My goal was to simplify ssh into the zd1200 by supporting the current client defaults, not to get rid of rsa to improve the minimum security of ssh sessions.
Yeah, I had to survive firmware upgrades, hence the nasty hack. Your way is much better.
 
  • Like
Reactions: dig

dig

New Member
May 13, 2021
20
20
3
Ruckus APs (at least those running Unleashed) do a heartbeat check by pinging "well known" servers on the Internet; if they can't reach any, they turn off wireless connectivity. How do you take care of this if the management VLAN does not have Internet access? I wonder if they try all VLANs.
I do not see any "Internet Status" indication on the ZD1200 interface, as I do with Unleashed. Even with the firewall set up to block the management VLAN access to the internet, the Unleashed status shows the internet is connected, which would fit with your hypothesis that it tries to access the internet through non-management VLANs. In both ZD and Unleased networks, the APs continue to provide wireless connectivity.
 

tgl

Member
Dec 23, 2024
60
27
18
Ruckus APs (at least those running Unleashed) do a heartbeat check by pinging "well known" servers on the Internet; if they can't reach any, they turn off wireless connectivity.
Really? Is that not configurable? That behavior would render them unfit for purpose, so far as I'm concerned. I don't want my local net to go down if my ISP has a glitch.
 

dig

New Member
May 13, 2021
20
20
3
Wireless mesh is not working with the virtual zd1200 and my R600 APs, I wonder if I can ask for some help to diagnose the issue with some captured DHCP packets from a working mesh setup running on a ZD1200, ideally version 10.5.1 and any model AP.

The behavior I see is the mesh AP associates and the WPA handshake completes, but it never pulls a DHCP address and ends up in SULKING. Captures show its DHCP discovers arriving at the root decrypted and well-formed, but they never make it to the root's mesh bridge.

If anyon has a working ZD-managed mesh running v 10.5.1, it's in the Troubleshooting->Diagnostics->Packet Capture menu item. Choose the wired/Root AP and unplug the other AP that connects via mesh. Start a "local" recording and then plug in the mesh AP and let it connect to the network and stabilize. then save the file and use these tshark commands to find a specific kind of packet and print out its raw content:

tshark -r mesh.pcap0 -Y "wlan.fc.type_subtype == 0x0028 && wlan.fc.ds == 0x03 && wlan.qos.amsdupresent == 0" | head
tshark -r mesh.pcap0 -Y "frame.number==N" -x

I just need the two hex lines around where the IP header starts (45 00) — specifically the four bytes before it. Mine read 81 00 00 01; my working Unleashed mesh reads 00 01 08 00, and I'm trying to find out which one a healthy ZD mesh produces. Let me know what model AP is involved, too. Many thanks!
 

unmesh

Active Member
Apr 17, 2017
280
74
28
67
Really? Is that not configurable? That behavior would render them unfit for purpose, so far as I'm concerned. I don't want my local net to go down if my ISP has a glitch.
That is how I discovered the "feature". I then configured pfSense to respond to pings to baidu.com!

@ms264556 points out that newer versions of Unleashed make this configurable
 

TonyArrr

Active Member
Sep 22, 2021
206
101
43
Straylia
Had a thought and was surprised not to see an option in Unleashed.
I was planning to link my neighbour and my networks for the purpose of sharing my LANcache (we play a lot of the same games, but I am blessed with a FTTP connection now and they aren’t), and was planning to just have my R750 joint their wifi as a client and firewall (do some DNS and static routing work on their network and so on).
I hadn’t gotten to it yet cause I’ve been rearranging my network segments and enabling IPv6 with the new ISP and so on, but just thought I’d see what I’d need to do on the AP and found that there is no option to join an existing network as a client in the WebUI at least.
Is it something possible to do on Unleashed? I’ve previously had Wifi APs as clients to a main wireless network to bridge wired clients before but this was a while before I had anything Ruckus, and I guess that is something you’d do less of in an enterprise setting… and if you did, you’d probably be picking a AP with more Ethernet ports
 

ms264556

Well-Known Member
Sep 13, 2021
522
504
93
New Zealand
ms264556.net
The nice thing to do (but lots more effort) would be to replace the sesame2 private key, so you can use the !v54! command to securely get a root shell. The key is plaintext inside the sesame executable, so this would be a safe thing to patch. If you get lots of people using your docker recipe, then maybe look at this.
sesame2 passphrases are keyed to the ZD serial number, so it's worth figuring out how to generate a random serial and apply as part of your setup.

Here is python code to patch sesame2 and give you a new passphrase. I didn't completely patch out the check: it could be worth keeping some security gate between the console and a root shell.

Bash:
    # 1) One-shot: re-key with a fresh random EC key + remove window check + print passphrase
python3 patch_sesame2.py <sesame2-bin> [out] --serial 123456789012
#    -> prints passphrase on stdout; saves new private key to <out>.key.pem

# 2) Keygen-only: mint a passphrase with an existing private key
python3 patch_sesame2.py --keygen-only 123456789012 --privkey <out>.key.pem [--verify]

# 3) Check-only: just remove the validity-window check
python3 patch_sesame2.py --check-only <sesame2-bin> [out]
Python:
#!/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()