IT_Ticket_System/app/services/license_service.py

"""
TechDesk License Service — offline RSA-signed key validation.
"""

import base64
import json
import logging
from datetime import date, datetime

logger = logging.getLogger(__name__)

PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxAMwjewlFaHpP+e69jM8
2I630u4B66kHcnqen2bxK65CZxvTBlLy2k/sm4zVVFyX0nf/9bDDio0UUTJrKWsp
kWV91Ag6FbvMgiJixfg5cZmpOd/abMs2OXAT/GZY+TLoszduchLLbkZjfDlhJzlt
D0MmAU9IY+zm9kW/YLWujFE37CFAfMLkDasQ7ZKKqjzQH6jOi8MoccI5zk8GlpoA
zlppk4dyQejmVDIXItg/0WVxaUa26tFafpU1RROlOKsw1lFXlWRQkc/uuJCdM943
PX0y1ZCCoNuOKofTIeBT506V7oM4g0rHGmD7esQJsz2S5E2rLetyLwcMKNjzo4a4
dwIDAQAB
-----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)