162 lines
5.1 KiB
Python
162 lines
5.1 KiB
Python
"""
|
|
License Check
|
|
=============
|
|
Validates the stored license key against Supabase on every launch.
|
|
Falls back to a local cache if the server is unreachable (30-minute grace period).
|
|
|
|
Storage location: %APPDATA%\StockScreener\
|
|
license.dat — XOR-encrypted license key (machine-bound)
|
|
auth_cache.dat — last successful auth result + timestamp
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
import requests
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Supabase config
|
|
# ---------------------------------------------------------------------------
|
|
_SUPABASE_URL = "https://yeispcpmepjelfbhfkwr.supabase.co"
|
|
_SUPABASE_ANON = (
|
|
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
|
|
"eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InllaXNwY3BtZXBqZWxmYmhma3dyIiwi"
|
|
"cm9sZSI6ImFub24iLCJpYXQiOjE3NzM1MDkxOTAsImV4cCI6MjA4OTA4NTE5MH0."
|
|
"8kq6HzpwF81_Sx36MXbx0wUqaT9Ul9POw1LzF5vxiOo"
|
|
)
|
|
|
|
# 30 minutes
|
|
_GRACE_PERIOD_SECONDS = 30 * 60
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths
|
|
# ---------------------------------------------------------------------------
|
|
_APP_DATA = os.path.join(os.environ.get("APPDATA", os.path.expanduser("~")), "StockScreener")
|
|
_LICENSE_FILE = os.path.join(_APP_DATA, "license.dat")
|
|
_CACHE_FILE = os.path.join(_APP_DATA, "auth_cache.dat")
|
|
|
|
|
|
def _ensure_dir() -> None:
|
|
os.makedirs(_APP_DATA, exist_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Encryption helpers (XOR cipher keyed on HWID — machine-binds the stored key)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _derive_key(hwid: str) -> bytes:
|
|
return hashlib.sha256(hwid.encode("utf-8")).digest()
|
|
|
|
|
|
def _xor(data: bytes, key: bytes) -> bytes:
|
|
return bytes(b ^ key[i % len(key)] for i, b in enumerate(data))
|
|
|
|
|
|
def _encrypt(text: str, hwid: str) -> str:
|
|
raw = _xor(text.encode("utf-8"), _derive_key(hwid))
|
|
return base64.urlsafe_b64encode(raw).decode("ascii")
|
|
|
|
|
|
def _decrypt(token: str, hwid: str) -> str:
|
|
raw = base64.urlsafe_b64decode(token.encode("ascii"))
|
|
return _xor(raw, _derive_key(hwid)).decode("utf-8")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# License key persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def save_license_key(key: str, hwid: str) -> None:
|
|
"""Encrypt and persist the license key to disk."""
|
|
_ensure_dir()
|
|
token = _encrypt(key, hwid)
|
|
with open(_LICENSE_FILE, "w", encoding="ascii") as f:
|
|
f.write(token)
|
|
|
|
|
|
def load_license_key(hwid: str) -> str | None:
|
|
"""Return the stored license key, or None if not found / corrupt."""
|
|
if not os.path.exists(_LICENSE_FILE):
|
|
return None
|
|
try:
|
|
with open(_LICENSE_FILE, "r", encoding="ascii") as f:
|
|
token = f.read().strip()
|
|
return _decrypt(token, hwid)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def delete_license_key() -> None:
|
|
"""Remove stored license key (e.g. on deactivation)."""
|
|
if os.path.exists(_LICENSE_FILE):
|
|
os.remove(_LICENSE_FILE)
|
|
if os.path.exists(_CACHE_FILE):
|
|
os.remove(_CACHE_FILE)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _save_cache(valid: bool, reason: str) -> None:
|
|
_ensure_dir()
|
|
payload = {"valid": valid, "reason": reason, "timestamp": time.time()}
|
|
with open(_CACHE_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f)
|
|
|
|
|
|
def _load_cache() -> dict | None:
|
|
if not os.path.exists(_CACHE_FILE):
|
|
return None
|
|
try:
|
|
with open(_CACHE_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core verification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def verify_license(license_key: str, hwid: str) -> tuple[bool, str]:
|
|
"""
|
|
Returns (valid: bool, reason: str).
|
|
|
|
Tries the Supabase RPC first. If the server is unreachable, falls back
|
|
to the cached result if it is within the 30-minute grace period.
|
|
"""
|
|
headers = {
|
|
"apikey": _SUPABASE_ANON,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(
|
|
f"{_SUPABASE_URL}/rest/v1/rpc/verify_license",
|
|
headers=headers,
|
|
json={"p_key": license_key, "p_hwid": hwid},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
valid = bool(data.get("valid", False))
|
|
reason = data.get("reason", "Unknown error")
|
|
_save_cache(valid, reason)
|
|
return valid, reason
|
|
|
|
except requests.RequestException:
|
|
cache = _load_cache()
|
|
if cache and cache.get("valid"):
|
|
age = time.time() - cache.get("timestamp", 0)
|
|
if age < _GRACE_PERIOD_SECONDS:
|
|
return True, "offline_grace"
|
|
return (
|
|
False,
|
|
"Cannot reach the license server. "
|
|
"Please check your internet connection and try again."
|
|
)
|