feat: implement C#-compatible AES-128-CBC license file parsing and integrate license management UI into AboutDialog
This commit is contained in:
jack
@@ -0,0 +1,15 @@
|
||||
from pygui.backend.license.license_manager import (
|
||||
LICENSE_FILENAME_RE,
|
||||
License,
|
||||
parse_license_blob,
|
||||
read_license_files,
|
||||
)
|
||||
from pygui.backend.license.license_model import LicenseModel
|
||||
|
||||
__all__ = [
|
||||
"LICENSE_FILENAME_RE",
|
||||
"License",
|
||||
"LicenseModel",
|
||||
"parse_license_blob",
|
||||
"read_license_files",
|
||||
]
|
||||
@@ -0,0 +1,146 @@
|
||||
"""QML-facing license model: grid rows, Load License, replay trial marker."""
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
from PySide6.QtCore import Property, QObject, Signal, Slot
|
||||
|
||||
from pygui.backend.license.license_manager import (
|
||||
LICENSE_FILENAME_RE,
|
||||
License,
|
||||
parse_license_blob,
|
||||
read_license_files,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
TRIAL_DAYS = 30
|
||||
|
||||
|
||||
def _to_local_path(path_or_url: str) -> Path:
|
||||
"""Accept both plain paths and file:// URLs (QML FileDialog gives URLs)."""
|
||||
if path_or_url.startswith("file:"):
|
||||
return Path(unquote(urlparse(path_or_url).path))
|
||||
return Path(path_or_url)
|
||||
|
||||
|
||||
# ===== License Model =====
|
||||
class LicenseModel(QObject):
|
||||
"""Context property backing the About dialog license grid.
|
||||
|
||||
Licenses live in ``<data_dir>/licenses`` (mirrors C#'s
|
||||
``DataDir\\licenses``). The 30-day replay trial marker is
|
||||
``<data_dir>/replay.lic``.
|
||||
"""
|
||||
|
||||
licensesChanged = Signal()
|
||||
|
||||
def __init__(self, data_dir: str, parent: QObject | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._data_dir = Path(data_dir)
|
||||
self._licenses_dir = self._data_dir / "licenses"
|
||||
try:
|
||||
self._licenses_dir.mkdir(parents=True, exist_ok=True)
|
||||
except OSError as exc:
|
||||
log.warning("Cannot create licenses dir %s: %s", self._licenses_dir, exc)
|
||||
self._licenses: list[License] = []
|
||||
self.refresh()
|
||||
|
||||
# ── Grid rows ─────────────────────────────────────────────────────
|
||||
@Property("QVariantList", notify=licensesChanged)
|
||||
def licenses(self) -> list:
|
||||
"""Rows for the About grid: serial, mfg date, level, license date."""
|
||||
return [
|
||||
{
|
||||
"serial": lic.serial,
|
||||
"mfgDate": lic.mfg_date_decimal,
|
||||
"level": lic.level,
|
||||
"licDate": lic.lic_date_decimal,
|
||||
}
|
||||
for lic in self._licenses
|
||||
]
|
||||
|
||||
@Slot()
|
||||
def refresh(self) -> None:
|
||||
"""Re-scan the licenses directory."""
|
||||
self._licenses = read_license_files(self._licenses_dir)
|
||||
self.licensesChanged.emit()
|
||||
|
||||
# ── Load License ──────────────────────────────────────────────────
|
||||
@Slot(str, result=bool)
|
||||
def loadLicenseFile(self, path_or_url: str) -> bool:
|
||||
"""Validate a picked .bin and copy it into the licenses directory."""
|
||||
src = _to_local_path(path_or_url)
|
||||
if not src.is_file():
|
||||
log.warning("License file does not exist: %s", src)
|
||||
return False
|
||||
if not LICENSE_FILENAME_RE.match(src.name):
|
||||
log.warning("License filename invalid: %s", src.name)
|
||||
return False
|
||||
|
||||
try:
|
||||
blob = src.read_bytes()
|
||||
except OSError as exc:
|
||||
log.warning("Cannot read license file %s: %s", src, exc)
|
||||
return False
|
||||
|
||||
lic = parse_license_blob(blob)
|
||||
if lic is None or src.name != f"{lic.serial}-{lic.level}.bin":
|
||||
log.warning("License file failed validation: %s", src.name)
|
||||
return False
|
||||
|
||||
try:
|
||||
self._licenses_dir.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(src, self._licenses_dir / src.name)
|
||||
except OSError as exc:
|
||||
log.warning("Cannot copy license into %s: %s", self._licenses_dir, exc)
|
||||
return False
|
||||
|
||||
self.refresh()
|
||||
return True
|
||||
|
||||
# ── Wafer detect (informational) ──────────────────────────────────
|
||||
@Slot(str, result=str)
|
||||
def levelForWafer(self, serial: str) -> str:
|
||||
"""License level for a wafer serial like "A00001"; "" if unlicensed."""
|
||||
for lic in self._licenses:
|
||||
if lic.serial == serial:
|
||||
return lic.level
|
||||
return ""
|
||||
|
||||
# ── Replay trial (30-day marker, C# parity) ───────────────────────
|
||||
@property
|
||||
def _trial_marker(self) -> Path:
|
||||
return self._data_dir / "replay.lic"
|
||||
|
||||
@Slot(result=str)
|
||||
def replayLicenseState(self) -> str:
|
||||
""""permanent" if any license level >= 1, "temporary" if trial
|
||||
marker exists, else "none" (trial not started)."""
|
||||
if any(lic.level_int >= 1 for lic in self._licenses):
|
||||
return "permanent"
|
||||
if self._trial_marker.exists():
|
||||
return "temporary"
|
||||
return "none"
|
||||
|
||||
@Slot(result=bool)
|
||||
def startReplayTrial(self) -> bool:
|
||||
"""Create the trial marker (idempotent). False on I/O failure."""
|
||||
try:
|
||||
self._trial_marker.touch(exist_ok=True)
|
||||
return True
|
||||
except OSError as exc:
|
||||
log.warning("Cannot create trial marker: %s", exc)
|
||||
return False
|
||||
|
||||
@Slot(result=bool)
|
||||
def replayTrialExpired(self) -> bool:
|
||||
"""True once the marker is older than 30 days (or missing)."""
|
||||
marker = self._trial_marker
|
||||
if not marker.exists():
|
||||
return True
|
||||
age_days = (time.time() - marker.stat().st_mtime) / 86400.0
|
||||
return age_days > TRIAL_DAYS
|
||||
@@ -0,0 +1,126 @@
|
||||
"""C#-compatible wafer license file parsing.
|
||||
|
||||
License files are AES-128-CBC encrypted `.bin` blobs named
|
||||
``X00000-00.bin`` where ``X`` is the wafer family code, ``00000`` the
|
||||
5-digit serial number, and ``00`` the 2-digit license level. The blob is
|
||||
``ciphertext || iv`` (IV is the trailing 16 bytes). The decrypted text is
|
||||
a fixed-offset record::
|
||||
|
||||
[0:6] serial e.g. "A00001"
|
||||
[7:15] mfg date 8 hex chars
|
||||
[16:18] level 2 decimal digits ("00" basic, "01"+ adds replay)
|
||||
[19:27] license date 8 hex chars
|
||||
|
||||
The key is hardcoded to match the C# application exactly
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from pygui.backend.crypto.crypto_helper import CryptoHelper
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
LICENSE_KEY = b"ThisIsA16ByteKey"
|
||||
LICENSE_FILENAME_RE = re.compile(r"^[A-Z]\d{5}-\d{2}\.bin$")
|
||||
|
||||
_MIN_TEXT_LEN = 27
|
||||
|
||||
|
||||
# ===== License Record =====
|
||||
@dataclass(frozen=True)
|
||||
class License:
|
||||
"""One parsed license file (all fields as stored in the file)."""
|
||||
|
||||
serial: str # "A00001"
|
||||
level: str # "00".."03"
|
||||
mfg_date_hex: str # 8 hex chars
|
||||
lic_date_hex: str # 8 hex chars
|
||||
|
||||
@property
|
||||
def level_int(self) -> int:
|
||||
"""Level as int; -1 if unparseable."""
|
||||
try:
|
||||
return int(self.level, 10)
|
||||
except ValueError:
|
||||
return -1
|
||||
|
||||
@property
|
||||
def mfg_date_decimal(self) -> str:
|
||||
"""Mfg date shown as decimal (C# grid does Convert.ToInt64(hex, 16))."""
|
||||
return _hex_to_decimal(self.mfg_date_hex)
|
||||
|
||||
@property
|
||||
def lic_date_decimal(self) -> str:
|
||||
"""License date shown as decimal, same conversion as the C# grid."""
|
||||
return _hex_to_decimal(self.lic_date_hex)
|
||||
|
||||
|
||||
def _hex_to_decimal(hex_str: str) -> str:
|
||||
try:
|
||||
return str(int(hex_str, 16))
|
||||
except ValueError:
|
||||
return ""
|
||||
|
||||
|
||||
# ===== Blob Parsing =====
|
||||
def split_data_iv(blob: bytes) -> tuple[bytes, bytes]:
|
||||
"""Split a license blob into (ciphertext, iv) — IV is the last 16 bytes."""
|
||||
if len(blob) < 32:
|
||||
raise ValueError("License blob too short")
|
||||
return blob[:-16], blob[-16:]
|
||||
|
||||
|
||||
def parse_license_blob(blob: bytes) -> License | None:
|
||||
"""Decrypt and parse one license blob. None if malformed."""
|
||||
try:
|
||||
data, iv = split_data_iv(blob)
|
||||
message = CryptoHelper.decrypt_aes(data, LICENSE_KEY, iv)
|
||||
text = message.decode("utf-8")
|
||||
except (ValueError, UnicodeDecodeError) as exc:
|
||||
log.warning("Undecryptable license blob: %s", exc)
|
||||
return None
|
||||
|
||||
if len(text) < _MIN_TEXT_LEN:
|
||||
log.warning("License text too short (%d chars)", len(text))
|
||||
return None
|
||||
|
||||
return License(
|
||||
serial=text[0:6],
|
||||
mfg_date_hex=text[7:15],
|
||||
level=text[16:18],
|
||||
lic_date_hex=text[19:27],
|
||||
)
|
||||
|
||||
|
||||
# ===== Directory Scan =====
|
||||
def read_license_files(licenses_dir: str | Path) -> list[License]:
|
||||
"""Parse every valid license .bin in the directory.
|
||||
|
||||
A file counts only if its name matches ``X00000-00.bin`` AND equals
|
||||
``{serial}-{level}.bin`` from its own decrypted payload — same
|
||||
tamper check as the C# app.
|
||||
"""
|
||||
directory = Path(licenses_dir)
|
||||
if not directory.is_dir():
|
||||
return []
|
||||
|
||||
licenses: list[License] = []
|
||||
for path in sorted(directory.glob("*.bin")):
|
||||
if not LICENSE_FILENAME_RE.match(path.name):
|
||||
continue
|
||||
try:
|
||||
blob = path.read_bytes()
|
||||
except OSError as exc:
|
||||
log.warning("Cannot read license file %s: %s", path, exc)
|
||||
continue
|
||||
lic = parse_license_blob(blob)
|
||||
if lic is None:
|
||||
continue
|
||||
if path.name != f"{lic.serial}-{lic.level}.bin":
|
||||
log.warning("License filename mismatch: %s", path.name)
|
||||
continue
|
||||
licenses.append(lic)
|
||||
return licenses
|
||||
Reference in New Issue
Block a user