ssz223
/
Stock-Tool
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Migrate app from Supabase/GitHub to Verimund API

This commit is contained in:
Nolan Kovacs 2026-03-25 20:38:32 -04:00
parent c36d4c2f03
commit 92c1750e1b
4 changed files with 975 additions and 409 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
api/main.py
View File

@ -116,6 +116,13 @@ def get_fundamentals_cache(payload=Depends(verify_jwt), db=Depends(get_db)):
return {"data": cur.fetchall()}
@app.get("/cache/sector")
def get_sector_stats(payload=Depends(verify_jwt), db=Depends(get_db)):
cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute("SELECT * FROM sector_stats ORDER BY sector")
return {"data": cur.fetchall()}
@app.get("/update")
def check_update(payload=Depends(verify_jwt), db=Depends(get_db)):
cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor)

1039
dist/stock_screener.py vendored
View File

File diff suppressed because it is too large Load Diff

119
license_check.py
View File

@ -1,16 +1,17 @@
"""
License Check
=============
Validates the stored license key against Supabase on every launch.
Validates the stored license key against the Verimund API 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
license.dat — XOR-encrypted license key (machine-bound)
auth_cache.dat — last successful auth result + timestamp + JWT token
"""
import base64
import hashlib
import hmac as _hmac_module
import json
import os
import time
@ -18,15 +19,13 @@ import time
import requests
# ---------------------------------------------------------------------------
# Supabase config
# API config
# ---------------------------------------------------------------------------
_SUPABASE_URL = "https://yeispcpmepjelfbhfkwr.supabase.co"
_SUPABASE_ANON = (
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
"eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InllaXNwY3BtZXBqZWxmYmhma3dyIiwi"
"cm9sZSI6ImFub24iLCJpYXQiOjE3NzM1MDkxOTAsImV4cCI6MjA4OTA4NTE5MH0."
"8kq6HzpwF81_Sx36MXbx0wUqaT9Ul9POw1LzF5vxiOo"
)
_API_URL = "https://api.verimundsolutions.com"
# HMAC secret — must match HMAC_SECRET in /srv/api/.env on the server.
# This value is obfuscated by PyArmor before distribution.
_HMAC_SECRET = "fa1fac57c9cc9742ac66542a17b032ccdfb2211b72c81fccd1617f17ce03ac7f"
# 30 minutes
_GRACE_PERIOD_SECONDS = 30 * 60
@ -101,21 +100,19 @@ def delete_license_key() -> None:
# Auth cache
# ---------------------------------------------------------------------------
def _save_cache(valid: bool, reason: str, channel: str = "stable") -> None:
def _save_cache(valid: bool, reason: str, channel: str = "stable", jwt_token: str = "") -> None:
_ensure_dir()
payload = {"valid": valid, "reason": reason, "channel": channel, "timestamp": time.time()}
payload = {
"valid": valid,
"reason": reason,
"channel": channel,
"jwt_token": jwt_token,
"timestamp": time.time(),
}
with open(_CACHE_FILE, "w", encoding="utf-8") as f:
json.dump(payload, f)
def get_channel() -> str:
"""Return the update channel ('stable' or 'beta') from the last auth cache."""
cache = _load_cache()
if cache:
return cache.get("channel", "stable")
return "stable"
def _load_cache() -> dict | None:
if not os.path.exists(_CACHE_FILE):
return None
@ -126,6 +123,43 @@ def _load_cache() -> dict | None:
return None
def get_channel() -> str:
"""Return the update channel ('stable' or 'beta') from the last auth cache."""
cache = _load_cache()
if cache:
return cache.get("channel", "stable")
return "stable"
def get_jwt() -> str:
"""Return the stored JWT token, or empty string if not available."""
cache = _load_cache()
if cache:
return cache.get("jwt_token", "")
return ""
def get_api_headers() -> dict:
"""Return headers for authenticated API requests."""
return {
"Authorization": f"Bearer {get_jwt()}",
"Content-Type": "application/json",
}
# ---------------------------------------------------------------------------
# HMAC request signing
# ---------------------------------------------------------------------------
def _sign_request(license_key: str, timestamp: str) -> str:
"""Generate HMAC-SHA256 signature for an auth request."""
return _hmac_module.new(
_HMAC_SECRET.encode(),
f"{license_key}{timestamp}".encode(),
hashlib.sha256,
).hexdigest()
# ---------------------------------------------------------------------------
# Core verification
# ---------------------------------------------------------------------------
@ -134,31 +168,41 @@ 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.
Calls POST /auth on the Verimund API with a signed request.
Falls back to cached result within the 30-minute grace period if offline.
"""
headers = {
"apikey": _SUPABASE_ANON,
"Content-Type": "application/json",
}
try:
timestamp = str(int(time.time()))
signature = _sign_request(license_key, timestamp)
resp = requests.post(
f"{_SUPABASE_URL}/rest/v1/rpc/verify_license",
headers=headers,
json={"p_key": license_key, "p_hwid": hwid},
f"{_API_URL}/auth",
json={
"license_key": license_key,
"hwid": hwid,
"timestamp": timestamp,
"signature": signature,
},
timeout=10,
)
resp.raise_for_status()
data = resp.json()
valid = bool(data.get("valid", False))
reason = data.get("reason", "Unknown error")
data = resp.json()
token = data.get("token", "")
channel = data.get("channel", "stable")
_save_cache(valid, reason, channel)
return valid, reason
_save_cache(True, "ok", channel, token)
return True, "ok"
except requests.exceptions.HTTPError as e:
try:
detail = e.response.json().get("detail", str(e))
except Exception:
detail = str(e)
_save_cache(False, detail, "stable", "")
return False, detail
except requests.exceptions.SSLError as e:
return False, f"SSL error contacting license server: {e}"
except requests.exceptions.ConnectionError as e:
cache = _load_cache()
if cache and cache.get("valid"):
@ -166,8 +210,7 @@ def verify_license(license_key: str, hwid: str) -> tuple[bool, str]:
if age < _GRACE_PERIOD_SECONDS:
return True, "offline_grace"
return False, f"Cannot connect to license server. Check your internet connection.\n\n({e})"
except requests.exceptions.HTTPError as e:
return False, f"License server error: {e}"
except requests.RequestException as e:
cache = _load_cache()
if cache and cache.get("valid"):

219
updater.py
View File

@ -1,11 +1,9 @@
"""
Auto-Updater
============
On every launch (after a successful license check), checks Supabase for a
newer version. If found:
- Downloads new screener_gui.py and stock_screener.py silently
- If a new exe is also available, self-replaces via a helper batch script
and restarts. The user sees nothing — the app just comes back up updated.
On every launch (after a successful license check), checks the Verimund API
for a newer version. If found, downloads the new exe via the API and
self-replaces via a helper batch script.
Called after a successful license check on every launch.
"""
@ -21,31 +19,7 @@ import requests
# ---------------------------------------------------------------------------
# Current app version — kept in sync by push_update.py before each build.
# ---------------------------------------------------------------------------
APP_VERSION = "1.3.1"
# ---------------------------------------------------------------------------
# GitHub token — fine-grained PAT with Contents:read on the release repo.
# Required to download assets from a private GitHub repository.
# Generate at: GitHub → Settings → Developer settings → Fine-grained tokens
# ---------------------------------------------------------------------------
_GITHUB_TOKEN = "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"
_GITHUB_REPO = "nolankovacs/ultimate-investment-tool"
# ---------------------------------------------------------------------------
# Supabase config
# ---------------------------------------------------------------------------
_SUPABASE_URL = "https://yeispcpmepjelfbhfkwr.supabase.co"
_SUPABASE_ANON = (
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
"eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InllaXNwY3BtZXBqZWxmYmhma3dyIiwi"
"cm9sZSI6ImFub24iLCJpYXQiOjE3NzM1MDkxOTAsImV4cCI6MjA4OTA4NTE5MH0."
"8kq6HzpwF81_Sx36MXbx0wUqaT9Ul9POw1LzF5vxiOo"
)
_HEADERS = {
"apikey": _SUPABASE_ANON,
"Content-Type": "application/json",
}
APP_VERSION = "1.3.6"
def _exe_dir() -> str:
@ -61,49 +35,59 @@ def _parse_version(v: str) -> tuple:
return (0, 0, 0)
def _download_file(url: str, dest_path: str) -> None:
"""Download url → dest_path atomically via a temp file."""
dl_headers = {}
if _GITHUB_TOKEN and "github.com" in url:
dl_headers["Authorization"] = f"Bearer {_GITHUB_TOKEN}"
# GitHub API asset endpoint requires this header to stream the binary
if "api.github.com" in url and "/releases/assets/" in url:
dl_headers["Accept"] = "application/octet-stream"
tmp_dir = tempfile.mkdtemp()
tmp_path = os.path.join(tmp_dir, os.path.basename(dest_path))
def _log_update_error(msg: str) -> None:
"""Log update errors silently — never let this crash the app."""
try:
with requests.get(url, stream=True, timeout=60, headers=dl_headers) as r:
r.raise_for_status()
with open(tmp_path, "wb") as f:
for chunk in r.iter_content(chunk_size=65536):
f.write(chunk)
shutil.move(tmp_path, dest_path)
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
from license_check import get_api_headers, _API_URL
from hwid import get_hwid
hwid = get_hwid()
import license_check
channel = license_check.get_channel()
requests.post(
f"{_API_URL}/log",
headers=get_api_headers(),
json={
"hwid": hwid,
"app_version": APP_VERSION,
"channel": channel,
"message": msg,
},
timeout=5,
)
except Exception:
pass
def _self_replace_exe(exe_url: str) -> None:
def _self_replace_exe(jwt_token: str, channel: str) -> None:
"""
Download the new exe to %TEMP%, then launch a helper batch script that
waits for this process to exit, moves the file into place, and restarts
the app. Nothing is left in the application directory. Never returns.
Download the new exe from the API, then launch a helper batch script that
waits for this process to exit, moves the file into place, and restarts.
Never returns.
"""
if not getattr(sys, "frozen", False):
return # only meaningful in a frozen exe
current_exe = os.path.abspath(sys.executable)
tmp_dir = tempfile.gettempdir()
pending_exe = os.path.join(tmp_dir, "StockScreener_pending.exe")
# Download to %TEMP% — nothing lands next to the running exe
try:
_download_file(exe_url, pending_exe)
except Exception as exc:
_log_update_error(f"_download_file failed for {exe_url}: {exc}")
return
# Batch script in %TEMP% — deletes itself and the pending file when done
from license_check import _API_URL
current_exe = os.path.abspath(sys.executable)
tmp_dir = tempfile.gettempdir()
pending_exe = os.path.join(tmp_dir, "StockScreener_pending.exe")
try:
with requests.post(
f"{_API_URL}/download",
headers={"Authorization": f"Bearer {jwt_token}", "Content-Type": "application/json"},
stream=True,
timeout=120,
) as r:
r.raise_for_status()
with open(pending_exe, "wb") as f:
for chunk in r.iter_content(chunk_size=65536):
f.write(chunk)
except Exception as exc:
_log_update_error(f"_self_replace_exe download failed: {exc}")
return
helper = os.path.join(tmp_dir, "uit_update.bat")
with open(helper, "w") as f:
f.write(
@ -116,7 +100,6 @@ def _self_replace_exe(exe_url: str) -> None:
f"del \"%~f0\"\n"
)
# Launch the helper detached, then exit so the running exe is unlocked
subprocess.Popen(
["cmd", "/c", helper],
creationflags=subprocess.CREATE_NO_WINDOW | subprocess.DETACHED_PROCESS,
@ -127,16 +110,14 @@ def _self_replace_exe(exe_url: str) -> None:
def get_release_notes() -> tuple:
"""
Returns (version, release_notes) for the current channel from Supabase.
Returns (version, release_notes) for the current channel from the API.
Falls back gracefully on any error.
"""
try:
import license_check
channel = license_check.get_channel()
resp = requests.post(
f"{_SUPABASE_URL}/rest/v1/rpc/get_latest_version",
headers=_HEADERS,
json={"p_channel": channel},
from license_check import get_api_headers, get_jwt, _API_URL
resp = requests.get(
f"{_API_URL}/update",
headers=get_api_headers(),
timeout=8,
)
resp.raise_for_status()
@ -148,108 +129,34 @@ def get_release_notes() -> tuple:
return APP_VERSION, "Unable to fetch release notes."
def _log_update_error(msg: str) -> None:
"""Upload update events/errors to Supabase update_logs table."""
# Remote upload only — no local file created
try:
hwid = "unknown"
channel = "unknown"
try:
from hwid import get_hwid
hwid = get_hwid()
except Exception:
pass
try:
import license_check
channel = license_check.get_channel()
except Exception:
pass
requests.post(
f"{_SUPABASE_URL}/rest/v1/update_logs",
headers={
**_HEADERS,
"Content-Type": "application/json",
"Prefer": "return=minimal",
},
json={
"hwid": hwid,
"app_version": APP_VERSION,
"channel": channel,
"message": msg,
},
timeout=5,
)
except Exception:
pass # never let logging break the app
def _get_github_exe_url(version: str, channel: str) -> str | None:
"""
Query the GitHub Releases API to find the StockScreener.exe asset URL
for the given version and channel. Used as a fallback when Supabase
doesn't return exe_url.
"""
try:
tag = f"v{version}-{channel}" if channel != "stable" else f"v{version}"
resp = requests.get(
f"https://api.github.com/repos/{_GITHUB_REPO}/releases/tags/{tag}",
headers={
"Authorization": f"Bearer {_GITHUB_TOKEN}",
"Accept": "application/vnd.github+json",
},
timeout=10,
)
resp.raise_for_status()
for asset in resp.json().get("assets", []):
if asset["name"] == "StockScreener.exe":
return asset["url"] # API asset URL — requires octet-stream header
_log_update_error(f"GitHub release {tag} found but no StockScreener.exe asset")
return None
except Exception as exc:
_log_update_error(f"_get_github_exe_url({version}, {channel}) failed: {exc}")
return None
def check_and_apply_update() -> None:
"""
Silently check for updates and apply them. Errors are swallowed so a
failed update check never blocks the app from launching.
"""
try:
import license_check
channel = license_check.get_channel()
from license_check import get_api_headers, get_jwt, get_channel, _API_URL
# Use the RPC — it runs with elevated DB permissions (SECURITY DEFINER)
# so the anon key can read the table through it even with RLS enabled.
resp = requests.post(
f"{_SUPABASE_URL}/rest/v1/rpc/get_latest_version",
headers=_HEADERS,
json={"p_channel": channel},
resp = requests.get(
f"{_API_URL}/update",
headers=get_api_headers(),
timeout=8,
)
resp.raise_for_status()
data = resp.json()
latest = data.get("version", APP_VERSION)
channel = data.get("channel", "stable")
if _parse_version(latest) <= _parse_version(APP_VERSION):
return # already up to date
# Always resolve the download URL via the GitHub API — direct
# github.com/releases/download URLs return 404 for private repos
# when accessed with a token outside of a browser session.
exe_url = _get_github_exe_url(latest, channel)
if not exe_url:
_log_update_error(
f"Update available ({APP_VERSION} → {latest}) but could not "
f"locate the asset on GitHub for channel={channel}"
)
jwt_token = get_jwt()
if not jwt_token:
_log_update_error("No JWT token available for update download")
return
# Self-replace exe if a new one is available (exits this process)
_self_replace_exe(exe_url)
_self_replace_exe(jwt_token, channel)
except Exception as exc:
_log_update_error(f"check_and_apply_update failed: {exc}")
# never crash the app over a failed update check