Skip to content

Instantly share code, notes, and snippets.

@LeHungryBoi

LeHungryBoi/gist:de133cb9746c80f4ff9cda5e6c61817a

Created July 5, 2026 09:34
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save LeHungryBoi/de133cb9746c80f4ff9cda5e6c61817a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save LeHungryBoi/de133cb9746c80f4ff9cda5e6c61817a to your computer and use it in GitHub Desktop.
Download ZIP
tier.py
Raw
gistfile1.txt
the tier authenticate server went down never to be seen again. I unpacked this file from the program. what do I do in a abandonware situation like this?
tier.py
# Server fucked emergency deactivation
# Internal filename: 'tier.py'
# Bytecode version: 3.10.b1 (3439)
# Source timestamp: 1970-01-01 00:00:00 UTC (0)
global _cached_machine_fingerprint
global _cached_token
global _cache_checked
# ***<module>: Failure: Different bytecode
"""License tier system with cryptographic activation.\n\nEach build ships with a TIER constant (\"free\", \"personal\", \"power\") that\ndetermines the base feature set. Personal and Power tiers require a license\nkey that is validated once online, producing a signed activation token stored\nlocally. The token is verified on every launch using the embedded public key —\nno internet required after activation.\n\nActivation token format (JSON, base64-encoded + signature):\n{\n \"tier\": \"personal\",\n \"key\": \"SS-XXXX-XXXX-XXXX\",\n \"machine\": \"<machine fingerprint>\",\n \"activated_at\": \"2026-04-13T...\",\n \"max_feeds\": 15,\n \"max_monitors\": 3,\n \"max_platforms\": 5,\n \"features\": {...}\n}\n\nThe token is signed with the server\'s Ed25519 private key. The app only has\nthe public key and can verify but never forge tokens.\n"""
# Server fucked emergency deactivation
# Server fucked emergency deactivation
# Server fucked emergency deactivation
import os
import sys
import json
import hashlib
import platform
import logging
import threading
from datetime import datetime, timezone
from typing import Optional
from version import APP_VERSION
logger = logging.getLogger(__name__)
from _build_tier import BUILD_TIER as TIER
try:
from _build_tier import _BUILD_PIPELINE_VERIFIED as _BUILD_VERIFIED
except ImportError:
_BUILD_VERIFIED = False
_tier_override = os.environ.get('STREAMSTASH_TIER_OVERRIDE', '').strip().lower()
if _tier_override and (not getattr(sys, 'frozen', False)):
if _tier_override in ['free', 'personal', 'power']:
logger.warning(f'Tier override via STREAMSTASH_TIER_OVERRIDE: {TIER!r} -> {_tier_override!r} (this process only)')
TIER = _tier_override
else:
logger.warning(f'Ignoring invalid STREAMSTASH_TIER_OVERRIDE: {_tier_override!r}')
if TIER not in ['free', 'personal', 'power']:
raise RuntimeError(f'Invalid BUILD_TIER in _build_tier.py: {TIER!r}')
else:
if getattr(sys, 'frozen', False) and (not _BUILD_VERIFIED):
raise RuntimeError('BUILD ERROR: _build_tier.py was not pinned by _prepare_build.py. Refusing to start to avoid shipping the dev-default tier. Use build.bat to produce a release build.')
else:
PUBLIC_KEY_HEX = 'ba189e718e827a8264b76edb87a41091a6d8493432ebdf0c3df43319a108cfb9'
ACTIVATION_URL = 'https://streamstash.live/api/activate'
def _app_data_dir() -> str:
if sys.platform == 'win32':
base = os.environ.get('APPDATA') or os.path.expanduser('~')
path = os.path.join(base, 'StreamStash')
else:
if sys.platform == 'darwin':
path = os.path.expanduser('~/Library/Application Support/StreamStash')
else:
path = os.path.expanduser('~/.config/streamstash')
os.makedirs(path, exist_ok=True)
return path
_TOKEN_FILE = os.path.join(_app_data_dir(), '.license')
TIER_DEFS = {'free': ['tiktok', 'twitter', 'youtube'], 'personal': 5, 'power': [1, 30, {'platforms': 'library', 'max_feeds': True, 'keyboard_shortcuts': True, 'dashboard': True, 'compression': False, 'bio_tracking': False, 'discord_webhooks': False, 'ai_search': False, 'hd_downloads': False, 'youtube_4k_oneoff': True, 'deduplication': False, 'identity_matching': False, 'telegram_chats': True, 'dashboard_analytics': True, 'ai_search': True, 'hd_downloads': True, 'youtube_4k_oneoff': True, 'deduplication': True, 'identity_matching': True, 'telegram_chats': True, 'dashboard_analytics': True, 'metric_insights': True, 'priority_support': True, 'auto_clipping': True}]}
_cached_machine_fingerprint: str | None = None
_machine_fingerprint_lock = threading.Lock()
def _bios_uuid_windows() -> str | None:
"""Query the BIOS/system UUID on Windows.\n Prefers PowerShell Get-CimInstance (Win11 24H2 compatible) and falls back\n to the deprecated wmic for older systems that lack CIM cmdlets."""
import subprocess
startupinfo = None
creationflags = 0
if sys.platform == 'win32':
CREATE_NO_WINDOW = 134217728
creationflags = CREATE_NO_WINDOW
try:
result = subprocess.run(['powershell', '-NoProfile', '-Command', '(Get-CimInstance Win32_ComputerSystemProduct).UUID'], capture_output=True, text=True, timeout=8, creationflags=creationflags, startupinfo=startupinfo)
uuid = result.stdout.strip()
if uuid and uuid.lower()!= 'ffffffff-ffff-ffff-ffff-ffffffffffff':
return uuid
except Exception as e:
logger.debug(f'Get-CimInstance fingerprint failed: {e}')
try:
result = subprocess.run(['wmic', 'csproduct', 'get', 'UUID'], capture_output=True, text=True, timeout=5, creationflags=creationflags, startupinfo=startupinfo)
lines = [l.strip() for l in result.stdout.strip().split('\n') if l.strip() and l.strip()!= 'UUID']
if lines:
return lines[0]
except Exception as e:
logger.debug(f'wmic fingerprint failed: {e}')
return None
def _machine_fingerprint() -> str:
"""Generate a stable machine fingerprint from hardware identifiers."""
global _cached_machine_fingerprint
if _cached_machine_fingerprint:
return _cached_machine_fingerprint
else:
with _machine_fingerprint_lock:
if _cached_machine_fingerprint:
return _cached_machine_fingerprint
else:
parts = []
if sys.platform == 'win32':
uuid = _bios_uuid_windows()
if uuid:
parts.append(uuid)
parts.append(platform.node())
parts.append(platform.machine())
raw = '|'.join(parts)
_cached_machine_fingerprint = hashlib.sha256(raw.encode()).hexdigest()[:32]
return _cached_machine_fingerprint
def _load_token() -> dict | None:
# irreducible cflow, using cdg fallback
"""Load and verify the stored activation token.\n Raises LicenseCryptoUnavailable if the build is missing its crypto library\n — a packaging bug that must not silently downgrade the user to free."""
# ***<module>._load_token: Failure: Compilation Error
if not os.path.exists(_TOKEN_FILE):
return None
with open(_TOKEN_FILE, 'r') as f:
data = json.load(f)
token_bytes = bytes.fromhex(data['token'])
signature = bytes.fromhex(data['signature'])
if not _verify_signature(token_bytes, signature):
logger.warning('License token signature invalid — activation required')
return
else:
token = json.loads(token_bytes.decode())
if token.get('machine')!= _machine_fingerprint():
logger.warning('License token machine mismatch — reactivation required')
return
else:
if token.get('tier')!= TIER:
logger.warning(f"License token tier ({token.get('tier')}) doesn\'t match build ({TIER})")
return
else:
expires_at_raw = token.get('expires_at')
if not expires_at_raw:
logger.warning('License token has no expiry — rejecting; reactivation required')
return
from datetime import datetime, timezone
expires_at = datetime.fromisoformat(expires_at_raw.replace('Z', '+00:00'))
if datetime.now(timezone.utc) >= expires_at:
logger.warning('License token expired — reactivation required')
return
token['_license_key'] = data.get('license_key')
return token
except (ValueError, TypeError) as e:
logger.warning(f'Could not parse token expiry {expires_at_raw!r}: {e}')
return None
except LicenseCryptoUnavailable:
raise
except Exception as e:
logger.warning(f'Failed to load license token: {e}')
return None
def _read_license_key_from_file() -> Optional[str]:
"""Read just the license_key from the .license file, independent of\n whether the signed token is still valid. Used when the token has\n expired but we want to attempt silent re-activation using the key."""
if not os.path.exists(_TOKEN_FILE):
return None
try:
with open(_TOKEN_FILE, 'r') as f:
data = json.load(f)
return data.get('license_key')
except Exception:
return None
def _revalidate_silently(license_key: str) -> tuple[bool, str]:
# irreducible cflow, using cdg fallback
"""Silently re-check the license with the activation server.\n\n Returns (still_valid, reason). Explicit 403 from the server (revoked,\n disabled, etc.) returns False. Network errors / server unreachable return\n True so offline users keep working with their cached token.\n """
# ***<module>._revalidate_silently: Failure: Compilation Error
# EMERGENCY BYPASS: activation server is down, skip all online checks
logger.warning('EMERGENCY BYPASS: skipping online revalidation (activation server unavailable)')
return (True, 'emergency-bypass')
def _save_token(token_bytes: bytes, signature: bytes, license_key: str=''):
"""Save the activation token to disk.\n\n license_key is stored in plain text alongside the signed token so the app\n can silently re-validate on startup without prompting the user. Safe to\n store locally — the token is already tied to this machine\'s fingerprint,\n so a copied .license file won\'t work on another machine anyway.\n """
payload = {'token': token_bytes.hex(), 'signature': signature.hex()}
if license_key:
payload['license_key'] = license_key
with open(_TOKEN_FILE, 'w') as f:
json.dump(payload, f)
class LicenseCryptoUnavailable(RuntimeError):
"""Raised when no Ed25519 crypto library is available.\n Missing the library is a packaging bug, not a bad-signature — this is\n distinct from \'token signature failed\' so callers can tell the user\n \'contact support, your build is broken\' instead of silently falling\n back to free tier."""
def _verify_signature(message: bytes, signature: bytes) -> bool:
# irreducible cflow, using cdg fallback
"""Verify an Ed25519 signature using the embedded public key.\n\n Raises LicenseCryptoUnavailable if neither PyNaCl nor cryptography is\n installed — this must be loud so a paying user doesn\'t silently\n downgrade to free after an incomplete install.\n """
# ***<module>._verify_signature: Failure: Compilation Error
from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError
vk = VerifyKey(bytes.fromhex(PUBLIC_KEY_HEX))
vk.verify(message, signature)
return True
except BadSignatureError:
return False
except ImportError:
pass
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature
pk = Ed25519PublicKey.from_public_bytes(bytes.fromhex(PUBLIC_KEY_HEX))
pk.verify(signature, message)
return True
except InvalidSignature:
return False
except ImportError:
raise LicenseCryptoUnavailable('Neither PyNaCl nor cryptography is installed — license verification cannot run. Reinstall StreamStash or `pip install pynacl`.')
def activate(license_key: str) -> tuple[bool, str]:
"""Activate a license key against the server.\n Returns (success, message)."""
global _cached_token
global _cache_checked
import requests
try:
resp = requests.post(ACTIVATION_URL, json={'key': license_key.strip(), 'machine': _machine_fingerprint(), 'tier': TIER, 'app_version': APP_VERSION}, timeout=15)
if resp.status_code!= 200:
error = resp.json().get('error', 'Activation failed') if resp.headers.get('content-type', '').startswith('application/json') else f'Server returned {resp.status_code}'
return (False, error)
else:
data = resp.json()
if not data.get('ok'):
return (False, data.get('error', 'Activation rejected by server'))
else:
token_bytes = bytes.fromhex(data['token'])
signature = bytes.fromhex(data['signature'])
if not _verify_signature(token_bytes, signature):
return (False, 'Token signature verification failed')
else:
token = json.loads(token_bytes.decode())
if token.get('tier')!= TIER:
return (False, f"License is for {token.get('tier')} tier, but this is a {TIER} build")
else:
if token.get('machine')!= _machine_fingerprint():
return (False, 'Activation token is bound to a different machine')
else:
_save_token(token_bytes, signature, license_key.strip())
with _cache_lock:
_cached_token = None
_cache_checked = False
logger.info(f'License activated: {TIER} tier, key={license_key[:8]}...')
return (True, 'Activated successfully')
except requests.ConnectionError:
return (False, 'Could not connect to activation server. Check your internet connection.')
except requests.Timeout:
return (False, 'Activation server timed out. Try again.')
except Exception as e:
return (False, f'Activation error: {e}')
def deactivate():
"""Remove the local activation token."""
global _cached_token
global _cache_checked
try:
if os.path.exists(_TOKEN_FILE):
os.remove(_TOKEN_FILE)
except Exception:
pass
with _cache_lock:
_cached_token = None
_cache_checked = False
_cached_token = None
_cache_checked = False
_cache_lock = threading.Lock()
def _get_active_tier() -> dict:
"""Get the active tier definition. For free tier, always returns the free def.\n For paid tiers, verifies the activation token first."""
global _cached_token
global _cache_checked
if TIER == 'free':
return TIER_DEFS['free']
else:
with _cache_lock:
if not _cache_checked:
_cached_token = _load_token()
license_key = _cached_token.get('_license_key') if _cached_token is not None else _read_license_key_from_file()
if license_key:
still_valid, reason = _revalidate_silently(license_key)
if not still_valid:
logger.warning(f'License no longer valid: {reason} — downgrading to free')
_cached_token = None
try:
os.remove(_TOKEN_FILE)
except Exception:
pass
else:
if reason == 'refreshed':
_cached_token = _load_token()
_cache_checked = True
has_token = _cached_token is not None
if has_token:
return TIER_DEFS.get(TIER, TIER_DEFS['free'])
else:
return TIER_DEFS['free']
def is_activated() -> bool:
"""Check if the current build is activated (or free tier)."""
if TIER == 'free':
return True
else:
_get_active_tier()
with _cache_lock:
return _cached_token is not None
def _drop_cached_token():
"""Clear the in-process token cache and delete the on-disk .license\n token. Called when the activation server tells us the license has been\n revoked (via the version-check piggyback in app.py). After this call,\n any subsequent has_feature / get_max_feeds / is_activated returns\n Free-tier results immediately — no restart required."""
global _cached_token
with _cache_lock:
_cached_token = None
try:
os.remove(_TOKEN_FILE)
except Exception:
return None
def get_tier_name() -> str:
"""Get the current active tier name."""
return TIER if is_activated() else 'free'
def get_max_feeds() -> int:
return _get_active_tier()['max_feeds']
def get_max_monitors() -> int:
return _get_active_tier()['max_monitors']
def get_update_delay_days() -> int:
"""Days to hold back automatic updates (Free tier gets 30, paid tiers 0)."""
return _get_active_tier().get('update_delay_days', 0)
def has_feature(feature: str) -> bool:
"""Check if a feature is available in the current tier."""
return _get_active_tier()['features'].get(feature, False)
def dedup_active() -> bool:
"""Dedup runs only when both the tier permits it AND the user hasn\'t\n disabled the Deduplication toggle in Settings → Features. Hashes are\n still stored regardless — only the skip-on-duplicate behavior is gated."""
if not _get_active_tier()['features'].get('deduplication', False):
return False
else:
try:
import db
return db.get_setting('feature_deduplication', '1')!= '0'
except Exception:
return True
def live_events_capture_active() -> bool:
"""Layer 1 gate — whether TikTok-Live chat / gift / poll events get\n captured to the DB during a recording. Power-tier eligibility (same\n tier flag as auto-clipping, since the data feeds both) plus the user\'s\n Settings → TikTok & Recordings → \'Capture chat & gifts\' toggle.\n\n Auto-clipping depends on this — no events means nothing to score —\n but the inverse is allowed: capture ON / auto-clip OFF gives a user\n the chat & gifts rail without producing clips."""
if not _get_active_tier()['features'].get('auto_clipping', False):
return False
else:
try:
import db
return db.get_setting('feature_live_events_capture', '1')!= '0'
except Exception:
return True
def auto_clipping_active() -> bool:
"""Layer 2 gate — whether clips are extracted from captured events.\n Requires live_events_capture_active() (capture must be on to have\n data to score) AND the user\'s auto-clipping toggle. Flipping either\n off mid-stream cleanly stops new clips."""
if not live_events_capture_active():
return False
else:
try:
import db
return db.get_setting('feature_auto_clipping', '1')!= '0'
except Exception:
return True
def get_all_features() -> dict:
"""Get all feature flags for the current tier."""
return dict(_get_active_tier()['features'])
def get_enabled_platforms() -> list[str]:
"""Get the list of platforms available for the current tier."""
return list(_get_active_tier()['platforms'])
def is_platform_enabled(platform: str) -> bool:
"""Check if a specific platform is available in the current tier."""
return platform in _get_active_tier()['platforms']
def can_add_feed() -> tuple[bool, str]:
"""Check if the user can add another feed. Returns (allowed, message)."""
import db
max_f = get_max_feeds()
with db.get_conn() as conn:
total = 0
for table in ['feed', 'instagram_feed', 'twitter_feed', 'telegram_feed', 'forum_thread', 'reddit_thread', 'snapchat_feed', 'album_feed']:
total += conn.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0]
if total >= max_f:
tier = get_tier_name()
return (False, f'Feed limit reached ({total}/{max_f}). Upgrade from {tier.title()} to add more.')
else:
return (True, '')
def can_add_monitor() -> tuple[bool, str]:
"""Check if the user can add another monitor. Returns (allowed, message)."""
import db
max_m = get_max_monitors()
with db.get_conn() as conn:
total = conn.execute('SELECT COUNT(*) FROM monitor').fetchone()[0]
if total >= max_m:
tier = get_tier_name()
return (False, f'Monitor limit reached ({total}/{max_m}). Upgrade from {tier.title()} to add more.')
else:
return (True, '')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment