Mi BL sniper: NTP-synced, latency-compensated midnight shot at Mi Community BL auth

__main__.py

#!/usr/bin/env python3
from enum import IntEnum
import datetime
from itertools import product
import os
import ntplib  # type: ignore
import multiprocessing
import random
import statistics
import time
from zoneinfo import ZoneInfo

import hashlib
import httpx

from rich.console import Console

# We use anycast DNS with low jitter, feel free to change this
NTP_SERVER = "time.cloudflare.com"

# For measuring latency to the API, should be enough for a stable P75
PROBE_COUNT = 20

# Match hardware threads - 1 (I have 8 cores)
FIRE_EARLY_OFFSETS_MS = [
    500,
    200,
    50,
    20,
    10,
    5,
    0,
]

TARGET_TZ = ZoneInfo("Asia/Shanghai")

APP_VERSION_CODE = "500411"
APP_VERSION_NAME = "5.4.11"

API_BASE = "" # you should know where to find this
STATUS_URL = f"{API_BASE}/user/bl-switch/state"
APPLY_URL = f"{API_BASE}/apply/bl-auth"

console = Console()


# --- Timing Utilities ---
def get_ntp(samples: int = 10) -> tuple[float, float]:
    """Returns seconds to add to `time.time()` to get NTP-corrected time."""
    client = ntplib.NTPClient()
    console.print(
        f"Syncing NTP from [bold cyan]{NTP_SERVER}[/bold cyan] ({samples} samples)..."
    )

    offsets = []
    for _ in range(samples):
        try:
            response = client.request(NTP_SERVER, version=4)
            offsets.append(response.offset)
            time.sleep(1)  # be polite to the server
        except ntplib.NTPException as e:
            console.print(f"[yellow]NTP sample failed: {e}[/yellow]")

    if not offsets:
        raise RuntimeError("Failed to get any NTP samples")

    offset = statistics.median(offsets)
    jitter = max(offsets) - min(offsets)

    console.print(
        "NTP:",
        f"offset [magenta]{offset * 1000:+.3f}ms[/magenta]",
        f"jitter [red]{jitter * 1000:.3f}ms[/red]",
        f"(median of {len(offsets)} samples)",
    )
    return offset, jitter


def get_fire_offsets(
    ntp_jitter: float, accounts: list, power: float = 2.0
) -> list[tuple[str, str, float]]:
    """Returns (token, device_id, offset_sec) for each process."""
    count = len(accounts)
    low = -ntp_jitter * 0.5
    high = ntp_jitter * 0.5

    jitter = [
        low + ((i / max(count - 1, 1)) ** power) * (high - low) for i in range(count)
    ]

    return [
        (token, device_id, jitter[i] - (early_ms / 1000.0))
        for (i, (token, device_id)), early_ms in product(
            enumerate(accounts), FIRE_EARLY_OFFSETS_MS
        )
    ]


def get_time_now(ntp_offset: float = 0) -> datetime.datetime:
    return datetime.datetime.fromtimestamp(time.time() + ntp_offset, tz=TARGET_TZ)


def get_next_midnight(ntp_offset: float) -> datetime.datetime:
    now = get_time_now(ntp_offset)
    tomorrow = now.date() + datetime.timedelta(days=1)
    return datetime.datetime.combine(tomorrow, datetime.time(0, 0, 0), tzinfo=TARGET_TZ)


def wait_until(target: datetime.datetime, ntp_offset: float = 0):
    now = get_time_now(ntp_offset)
    if target <= now:
        console.print(f"[yellow]{target} is in the past. Skipping wait.[/yellow]")
        return

    delay = (target - now).total_seconds()
    console.print(f"Sleeping for [cyan]{delay:.3f}s[/cyan] until {target}...")

    while True:
        now = get_time_now(ntp_offset)
        remaining = (target - now).total_seconds()
        if remaining <= 0:
            break
        time.sleep(min(remaining, 1.0))  # save system resources


# --- API Utilities ---
def generate_device_id() -> str:
    raw = f"{random.random()}-{time.time()}"
    return hashlib.sha1(raw.encode()).hexdigest().upper()


def build_cookie(token: str, device_id: str) -> str:
    return (
        f"new_bbs_serviceToken={token};"
        f"versionCode={APP_VERSION_CODE};"
        f"versionName={APP_VERSION_NAME};"
        f"deviceId={device_id};"
    )


def build_client(cookie: str) -> httpx.Client:
    return httpx.Client(
        headers={
            "User-Agent": "okhttp/4.12.0",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
            "Cookie": cookie,
        },
        timeout=httpx.Timeout(connect=2.0, read=15.0, write=5.0, pool=5.0),
        http1=True,
    )


def measure_latency(client: httpx.Client, url: str, count: int = PROBE_COUNT) -> float:
    """Returns P75 HTTP latency in seconds against the target host."""
    console.print(
        f"Probing latency to [bold cyan]{url}[/bold cyan] ({count} samples)..."
    )

    samples = []
    # Establishes TCP+TLS, don't count it
    try:
        client.head(url, timeout=5)
    except Exception:
        pass

    for _ in range(count):
        try:
            t0 = time.perf_counter()
            client.head(url, timeout=5)
            t1 = time.perf_counter()
            samples.append(t1 - t0)
            time.sleep(0.1)
        except Exception as e:
            console.print(f"[yellow]Probe failed: {e}[/yellow]")

    if not samples:
        raise RuntimeError("All latency probes failed")

    samples.sort()
    p75 = statistics.quantiles(samples, n=4)[2]  # 75th percentile

    console.print(
        "Latency:",
        f"min [green]{min(samples) * 1000:.3f}ms[/green]",
        f"p75 [bold magenta]{p75 * 1000:.3f}ms[/bold magenta]",
        f"max [red]{max(samples) * 1000:.3f}ms[/red]",
    )
    return p75  # add ntp_jitter budget on top


class ApiCode(IntEnum):
    SUCCESS = 0
    UNAUTHORIZED = 100001
    UNKNOWN = 100003
    EXPIRED = 100004


def raise_for_api_code(code: int):
    match code:
        case ApiCode.SUCCESS:
            return
        case ApiCode.UNAUTHORIZED:
            console.print("[red]Unauthorized. Please check your token.[/red]")
            raise RuntimeError("Unauthorized")
        case ApiCode.UNKNOWN:
            # server-side uncertain, treat as possible success
            console.print("[yellow]Abiguous status. Verify.[/yellow]")
            return
        case ApiCode.EXPIRED:
            console.print(
                "[red]Token is expired. Please refresh your token and try again.[/red]"
            )
            raise RuntimeError("Token expired")
        case _:
            console.print(
                f"[yellow]Unrecognized API code: {code}. Proceeding cautiously.[/yellow]"
            )


# -- API ---
class UnlockIsPass(IntEnum):
    APPROVED = 1
    ELIGIBLE = 4


class UnlockButtonState(IntEnum):
    READY = 1
    BLOCKED = 2
    TOO_NEW = 3


def check_unlock_availability(
    client: httpx.Client, raise_on_status: bool = True
) -> bool:
    """Checks if the account is eligible and ready to apply for unlocking. Returns True if ready, False otherwise."""
    response = client.get(STATUS_URL)
    if raise_on_status:
        response.raise_for_status()
    else:
        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            console.print("[yellow]Status check HTTP error:[/yellow]", e)
            return False

    data = response.json()
    raise_for_api_code(data["code"])
    data = data["data"]

    is_pass = data["is_pass"]
    button = data["button_state"]
    deadline = data.get("deadline_format", "N/A")

    match is_pass:
        case UnlockIsPass.APPROVED:
            console.print("[bold green]Request already approved![/bold green]")
            console.print(
                f"[green]Unlocking is possible until[green] [bold cyan]{deadline}[/bold cyan]"
            )
            return False
        case UnlockIsPass.ELIGIBLE:
            pass
        case _:
            raise RuntimeError(f"Unhandled is_pass: {is_pass}")

    match button:
        case UnlockButtonState.READY:
            console.print("[yellow]Ready to send request.[/yellow]")
            return True
        case UnlockButtonState.BLOCKED:
            console.print(
                f"[red]Unlocking is blocked until[/red] [bold cyan]{deadline}[/bold cyan]"
            )
            return False
        case UnlockButtonState.TOO_NEW:
            console.print("[red]Account is less than 30 days old.[/red]")
            return False
        case _:
            raise RuntimeError(f"Unhandled button_state: {button}")


def is_unlock_approved(client: httpx.Client) -> bool:
    """Returns True only if the status endpoint reports the request as approved."""
    try:
        response = client.get(STATUS_URL)
        response.raise_for_status()
        data = response.json()
    except (httpx.HTTPError, ValueError) as e:
        console.print("[yellow]Status verify failed:[/yellow]", e)
        return False

    if data.get("code") != ApiCode.SUCCESS:
        return False

    if data["data"].get("is_pass") == UnlockIsPass.APPROVED:
        deadline = data["data"].get("deadline_format", "N/A")
        console.print("[bold green]Verified: unlock approved![/bold green]")
        console.print(
            f"[green]Unlocking is possible until[/green] [bold cyan]{deadline}[/bold cyan]"
        )
        return True

    return False


class ApplyResult(IntEnum):
    APPROVED = 1
    REJECTED = 2
    LIMIT_REACHED = 3
    BLOCKED = 4


def request_unlock(client: httpx.Client, raise_on_status: bool = False) -> bool:
    """Sends the unlock request. Returns True if approved, False otherwise."""
    response = client.post(APPLY_URL, content=b'{"is_retry":true}')
    if raise_on_status:
        response.raise_for_status()
    else:
        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            console.print("[yellow]Status check HTTP error:[/yellow]", e)
            return False

    data = response.json()
    code = data["code"]

    if code == ApiCode.UNKNOWN:
        console.print(
            "[yellow]Ambiguous response (100003). Checking status...[/yellow]"
        )
        console.print(f"[dim]{data}[/dim]")
        return is_unlock_approved(client)

    raise_for_api_code(code)

    result = data["data"].get("apply_result")
    deadline = data["data"].get("deadline_format", "N/A")

    match result:
        case ApplyResult.APPROVED:
            console.print("[bold green]Unlock approved![/bold green]")
            return True
        case ApplyResult.REJECTED:
            console.print("[red]Request rejected.[/red]")
            console.print(f"[dim]{data}[/dim]")
            return False
        case ApplyResult.LIMIT_REACHED:
            console.print(
                f"[red]Daily limit reached. Try again after[/red] [orange]{deadline}[/orange]"
            )
            return False
        case ApplyResult.BLOCKED:
            console.print(
                f"[red]Account blocked until[/red] [orange]{deadline}[/orange]"
            )
            return False
        case _:
            console.print(f"[yellow]Unrecognized apply_result: {result}[/yellow]")
            console.print(f"[dim]{data}[/dim]")
            return False


# --- Main Logic ---
def filter_eligibility(accounts: list[tuple[str, str]]) -> list[tuple[str, str]]:
    eligible = []
    for token, device_id in accounts:
        with build_client(build_cookie(token, device_id)) as client:
            if check_unlock_availability(client):
                eligible.append((token, device_id))
    console.print(
        f"[bold green]{len(eligible)}[/bold green]/[bold cyan]{len(accounts)}[/bold cyan]",
        "accounts are eligible and ready.",
    )
    return eligible


def countdown(
    accounts: list[tuple[str, str]],
    ntp_offset: float,
    ntp_jitter: float,
    target: datetime.datetime,
):
    t60 = target - datetime.timedelta(seconds=60)
    t30 = target - datetime.timedelta(seconds=30)
    t10 = target - datetime.timedelta(seconds=10)

    # T-60s: fresh NTP
    wait_until(t60, ntp_offset)
    if (
        get_time_now(ntp_offset) >= t60
    ):  # don't bother refreshing if we're already within 30s of the target
        ntp_offset, ntp_jitter = get_ntp()
        console.print()
        console.print("[bold]NTP refreshed.[/bold]")

    # T-30s: latency probe + status check per account
    wait_until(t30, ntp_offset)
    console.print()
    with build_client(build_cookie(*accounts[0])) as probe_client:
        latency = measure_latency(probe_client, APPLY_URL)
    lead_time = latency + ntp_jitter
    console.print(f"Lead time: [magenta]{lead_time * 1000:.3f}ms[/magenta]")

    accounts = filter_eligibility(accounts)
    if not accounts:
        console.print("[red]No eligible accounts. Aborting.[/red]")
        raise RuntimeError("No eligible accounts.")

    # T-10s: handshake
    wait_until(t10, ntp_offset)
    console.print()

    fire_jobs = get_fire_offsets(ntp_jitter, accounts)
    console.print(f"[bold green]{len(fire_jobs)} processes ready to fire![/bold green]")

    barrier = multiprocessing.Barrier(len(fire_jobs))
    processes = []
    for i, (token, device_id, offset_sec) in enumerate(fire_jobs):
        core_id = (i % 7) + 1

        p = multiprocessing.Process(
            target=fire_worker,
            args=(
                barrier,
                core_id,
                token,
                device_id,
                target,
                lead_time,
                offset_sec,
                ntp_offset,
            ),
        )
        processes.append(p)

    for p in processes:
        p.start()
    for p in processes:
        p.join()


def fire_worker(
    barrier,
    core_id: int,
    token: str,
    device_id: str,
    target: datetime.datetime,
    lead_time: float,
    offset_sec: float,
    ntp_offset: float,
):
    os.sched_setaffinity(0, {core_id})

    worker_console = Console()
    worker_console.print(
        f"[dim]Process for offset {offset_sec * 1000:+.0f}ms pinned to Core {core_id}[/dim]"
    )

    fire_at = target + datetime.timedelta(seconds=offset_sec)
    sleep_until = fire_at - datetime.timedelta(seconds=lead_time)

    fire_at_timestamp = fire_at.timestamp()
    fire_at_timestamp_offsetted = fire_at_timestamp - ntp_offset
    # don't set this too big or it's literally DDoS-ing and I'm gonna kill you
    fire_at_timestamp_offsetted_redo = fire_at_timestamp_offsetted + 0.5

    offset_print = f"{offset_sec * 1000:+.3f}ms"

    with build_client(build_cookie(token, device_id)) as client:
        # pre-warm TCP+TLS
        try:
            client.head(APPLY_URL, timeout=3)
        except Exception:
            pass

        barrier.wait()

        # sleep until just before fire window
        wait_until(sleep_until, ntp_offset)

        now_time = time.time()
        now_perf = time.perf_counter()

        departure_timestamp = fire_at_timestamp_offsetted - lead_time

        seconds_remaining = departure_timestamp - now_time
        spin_target_perf = now_perf + seconds_remaining

        # busy-wait spin
        while time.perf_counter() < spin_target_perf:
            pass

        # fire!!!
        result = request_unlock(client)
        worker_console.print(f"[cyan]Fired (offset {offset_print})[/cyan]")

        while not result:
            result = request_unlock(client)
            worker_console.print(f"[cyan]Re-fired (offset {offset_print})[/cyan]")
            if time.time() > fire_at_timestamp_offsetted_redo:
                break


def main():
    ntp_offset, ntp_jitter = get_ntp()
    ntp_now = get_time_now(ntp_offset)

    with open("token.txt", "r") as f:
        tokens = [t for t in f.read().splitlines() if t.strip()]

    if not tokens:
        console.print("[red]No tokens found in [bold]token.txt[/bold]![/red]")
        raise RuntimeError("No tokens.")

    accounts = [(token, generate_device_id()) for token in tokens]
    accounts = filter_eligibility(accounts)
    if not accounts:
        console.print("[red]No eligible accounts. Aborting.[/red]")
        raise RuntimeError("No eligible accounts.")

    with build_client(build_cookie(*accounts[0])) as client:
        latency = measure_latency(client, APPLY_URL)
        lead_time = latency + ntp_jitter  # estimated, compute later

    console.print()
    console.print(f"NTP-corrected time: {ntp_now}")
    console.print(
        f"Estimated lead time (latency + NTP jitter): [bold magenta]{lead_time * 1000:.3f}ms[/bold magenta]"
    )
    console.print()

    console.print("[green]All systems nominal.[/green]")
    console.print("[bold red]Do not close this window![/bold red]")
    console.print()

    next_midnight = get_next_midnight(ntp_offset)  # next 00:00:00 UTC+8
    # next_midnight_minus_5sec = next_midnight - datetime.timedelta(seconds=5)

    # debug: wait for 1 minute from now
    # countdown(
    #     accounts, ntp_offset, ntp_jitter, target=ntp_now + datetime.timedelta(minutes=1)
    # )

    countdown(accounts, ntp_offset, ntp_jitter, target=next_midnight)


if __name__ == "__main__":
    main()