05/22 Implement license version
This commit is contained in:
@@ -0,0 +1,177 @@
|
||||
"""
|
||||
TechDesk License Service — offline RSA-signed key validation.
|
||||
|
||||
Key format: TDESK-<payload_b64>.<signature_b64>
|
||||
|
||||
payload_b64 = base64url( UTF-8 JSON bytes )
|
||||
signature_b64 = base64url( RSA-SHA256 signature of those bytes )
|
||||
|
||||
JSON payload fields:
|
||||
customer — company name
|
||||
email — contact email
|
||||
tier — "community" | "business" | "enterprise"
|
||||
issued_at — ISO date (YYYY-MM-DD)
|
||||
expires_at — ISO date (YYYY-MM-DD)
|
||||
|
||||
The public key below is the matching counterpart to the private key in
|
||||
license_server/private_key.pem. Replace the placeholder with the real key
|
||||
after running license_server/generate_keys.py for the first time.
|
||||
|
||||
Public key replacement steps:
|
||||
1. cd license_server && python generate_keys.py
|
||||
2. Copy the full contents of license_server/public_key.pem
|
||||
3. Replace the PUBLIC_KEY_PEM constant below with the copied text
|
||||
4. Restart the TechDesk app
|
||||
|
||||
The public key can only verify signatures — it cannot forge them.
|
||||
It is safe to embed in the application and ship to customers.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from datetime import date, datetime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Replace this with the output of license_server/generate_keys.py ──────────
|
||||
# After running generate_keys.py, copy the full contents of public_key.pem here.
|
||||
PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
|
||||
REPLACE_WITH_REAL_PUBLIC_KEY_FROM_license_server/generate_keys.py
|
||||
-----END PUBLIC KEY-----"""
|
||||
|
||||
# ── Tier → feature set mapping ───────────────────────────────────────────────
|
||||
TIER_FEATURES: dict[str, set] = {
|
||||
'community' : set(),
|
||||
'business' : {'chatbot', 'sla', 'digest', 'watchers', 'time_tracking', 'surveys'},
|
||||
'enterprise': {'chatbot', 'sla', 'digest', 'watchers', 'time_tracking', 'surveys',
|
||||
'email_ingestion'},
|
||||
}
|
||||
|
||||
_COMMUNITY_STATUS = {
|
||||
'valid' : False,
|
||||
'tier' : 'community',
|
||||
'customer' : None,
|
||||
'email' : None,
|
||||
'issued_at' : None,
|
||||
'expires_at': None,
|
||||
'days_left' : None,
|
||||
}
|
||||
|
||||
|
||||
def validate_license(key_string: str) -> dict:
|
||||
"""Verify an RSA-signed TechDesk license key and return a status dict.
|
||||
|
||||
Returns a dict with keys:
|
||||
valid bool
|
||||
tier str ('community' | 'business' | 'enterprise')
|
||||
customer str | None
|
||||
email str | None
|
||||
issued_at str | None (YYYY-MM-DD)
|
||||
expires_at str | None (YYYY-MM-DD)
|
||||
days_left int | None (None when invalid/no key)
|
||||
reason str (human-readable; present when valid=False)
|
||||
"""
|
||||
if not key_string or not key_string.strip():
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'no_key'}
|
||||
|
||||
key_string = key_string.strip()
|
||||
|
||||
if not key_string.startswith('TDESK-'):
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'invalid_format'}
|
||||
|
||||
# Check if the public key is still a placeholder
|
||||
if 'REPLACE_WITH_REAL_PUBLIC_KEY' in PUBLIC_KEY_PEM:
|
||||
logger.warning('[LICENSE] Public key is a placeholder — all keys will fail validation. '
|
||||
'Run license_server/generate_keys.py and update PUBLIC_KEY_PEM.')
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'no_public_key'}
|
||||
|
||||
try:
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
from cryptography.exceptions import InvalidSignature
|
||||
|
||||
raw = key_string[len('TDESK-'):]
|
||||
if '.' not in raw:
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'invalid_format'}
|
||||
|
||||
payload_b64, sig_b64 = raw.split('.', 1)
|
||||
payload_bytes = base64.urlsafe_b64decode(_pad_b64(payload_b64))
|
||||
signature = base64.urlsafe_b64decode(_pad_b64(sig_b64))
|
||||
|
||||
public_key = serialization.load_pem_public_key(PUBLIC_KEY_PEM.strip().encode())
|
||||
public_key.verify(signature, payload_bytes, padding.PKCS1v15(), hashes.SHA256())
|
||||
|
||||
payload = json.loads(payload_bytes.decode('utf-8'))
|
||||
expires_at = payload.get('expires_at', '')
|
||||
expires = date.fromisoformat(expires_at)
|
||||
today = date.today()
|
||||
|
||||
if expires < today:
|
||||
return {
|
||||
**_COMMUNITY_STATUS,
|
||||
'expires_at': expires_at,
|
||||
'customer' : payload.get('customer'),
|
||||
'email' : payload.get('email'),
|
||||
'issued_at' : payload.get('issued_at'),
|
||||
'reason' : 'expired',
|
||||
}
|
||||
|
||||
days_left = (expires - today).days
|
||||
tier = payload.get('tier', 'community')
|
||||
if tier not in TIER_FEATURES:
|
||||
tier = 'community'
|
||||
|
||||
logger.info(
|
||||
f'[LICENSE] Valid {tier} license for {payload.get("customer")} '
|
||||
f'(expires {expires_at}, {days_left} days left)'
|
||||
)
|
||||
return {
|
||||
'valid' : True,
|
||||
'tier' : tier,
|
||||
'customer' : payload.get('customer'),
|
||||
'email' : payload.get('email'),
|
||||
'issued_at' : payload.get('issued_at'),
|
||||
'expires_at': expires_at,
|
||||
'days_left' : days_left,
|
||||
}
|
||||
|
||||
except InvalidSignature:
|
||||
logger.warning('[LICENSE] Key signature verification failed — key may be tampered.')
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'invalid_signature'}
|
||||
except Exception as exc:
|
||||
logger.warning(f'[LICENSE] Key validation error: {exc}')
|
||||
return {**_COMMUNITY_STATUS, 'reason': 'invalid_format'}
|
||||
|
||||
|
||||
def get_status() -> dict:
|
||||
"""Return the cached license status stored in Flask app.config.
|
||||
|
||||
Must be called within an active Flask application context.
|
||||
Falls back to community status if not set (e.g. during testing).
|
||||
"""
|
||||
from flask import current_app
|
||||
return current_app.config.get('LICENSE_STATUS', {**_COMMUNITY_STATUS, 'reason': 'no_key'})
|
||||
|
||||
|
||||
def feature_enabled(feature_name: str) -> bool:
|
||||
"""Return True if the current license tier grants access to feature_name.
|
||||
|
||||
Must be called within an active Flask application context.
|
||||
"""
|
||||
status = get_status()
|
||||
tier = status.get('tier', 'community')
|
||||
return feature_name in TIER_FEATURES.get(tier, set())
|
||||
|
||||
|
||||
def days_until_expiry() -> 'int | None':
|
||||
"""Return days remaining on the current license, or None if invalid/community."""
|
||||
status = get_status()
|
||||
if not status.get('valid'):
|
||||
return None
|
||||
return status.get('days_left')
|
||||
|
||||
|
||||
def _pad_b64(s: str) -> str:
|
||||
"""Add base64url padding so Python's decoder accepts it."""
|
||||
return s + '=' * (-len(s) % 4)
|
||||
Reference in New Issue
Block a user