app.py

"""
============================================================
  LICENSE SERVER — Deploy this on a SEPARATE HuggingFace Space
  
  This server:
   - Stores valid license keys (in licenses.json)
   - Validates keys submitted by clients
   - Lets you add/revoke keys via admin API
   
  Set the env var ADMIN_SECRET in your HF Space secrets.
============================================================
"""

import os
import json
import uuid
import hashlib
from datetime import datetime, timezone
from pathlib import Path
from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import JSONResponse
import uvicorn

# ============================================================
# CONFIG
# ============================================================
ADMIN_SECRET = os.environ.get("ADMIN_SECRET", "change-me-super-secret")
LICENSE_FILE = Path("licenses.json")
PORT = int(os.environ.get("PORT", 7860))

# ============================================================
# LICENSE STORAGE
# ============================================================

def load_licenses() -> dict:
    """Load licenses from file."""
    if LICENSE_FILE.exists():
        try:
            return json.loads(LICENSE_FILE.read_text())
        except Exception:
            return {}
    return {}

def save_licenses(data: dict):
    """Save licenses to file."""
    LICENSE_FILE.write_text(json.dumps(data, indent=2))

def hash_key(key: str) -> str:
    """Hash a license key for storage (never store raw keys)."""
    return hashlib.sha256(key.encode()).hexdigest()

# ============================================================
# FASTAPI APP
# ============================================================
app = FastAPI(title="License Server", docs_url=None, redoc_url=None)

# ---- Admin auth dependency ----
def require_admin(x_admin_secret: str = Header(...)):
    if x_admin_secret != ADMIN_SECRET:
        raise HTTPException(status_code=403, detail="Invalid admin secret")

# ============================================================
# PUBLIC ENDPOINT — clients call this to validate their key
# ============================================================
@app.get("/validate")
async def validate_license(key: str):
    """
    Validate a license key.
    Returns {"valid": true/false, "message": "..."}
    """
    licenses = load_licenses()
    key_hash = hash_key(key)
    
    if key_hash not in licenses:
        return JSONResponse({"valid": False, "message": "Invalid license key."})
    
    entry = licenses[key_hash]
    
    # Check if revoked
    if entry.get("revoked"):
        return JSONResponse({"valid": False, "message": "License key has been revoked."})
    
    # Check expiry
    expires_at = entry.get("expires_at")
    if expires_at:
        expiry_dt = datetime.fromisoformat(expires_at)
        if datetime.now(timezone.utc) > expiry_dt:
            return JSONResponse({"valid": False, "message": "License key has expired."})
    
    # Update last_seen
    entry["last_seen"] = datetime.now(timezone.utc).isoformat()
    licenses[key_hash] = entry
    save_licenses(licenses)
    
    return JSONResponse({
        "valid": True,
        "message": "License key is valid.",
        "label": entry.get("label", ""),
        "expires_at": expires_at
    })

# ============================================================
# ADMIN ENDPOINTS — only you can call these
# ============================================================

@app.post("/admin/generate", dependencies=[Depends(require_admin)])
async def generate_key(label: str = "", expires_days: int = 0):
    """
    Generate a new license key.
    - label: optional note (e.g. customer name)
    - expires_days: 0 = never expires
    """
    key = str(uuid.uuid4()).replace("-", "").upper()
    key_formatted = f"LMA-{key[:8]}-{key[8:16]}-{key[16:24]}"
    key_hash = hash_key(key_formatted)
    
    expires_at = None
    if expires_days > 0:
        from datetime import timedelta
        expires_at = (datetime.now(timezone.utc) + timedelta(days=expires_days)).isoformat()
    
    licenses = load_licenses()
    licenses[key_hash] = {
        "label": label,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "expires_at": expires_at,
        "revoked": False,
        "last_seen": None
    }
    save_licenses(licenses)
    
    return {"key": key_formatted, "label": label, "expires_at": expires_at}

@app.post("/admin/revoke", dependencies=[Depends(require_admin)])
async def revoke_key(key: str):
    """Revoke a license key."""
    licenses = load_licenses()
    key_hash = hash_key(key)
    
    if key_hash not in licenses:
        raise HTTPException(status_code=404, detail="Key not found.")
    
    licenses[key_hash]["revoked"] = True
    save_licenses(licenses)
    return {"message": f"Key revoked successfully."}

@app.get("/admin/list", dependencies=[Depends(require_admin)])
async def list_keys():
    """List all license keys (hashed, with metadata)."""
    licenses = load_licenses()
    return {"total": len(licenses), "licenses": licenses}

@app.get("/health")
async def health():
    return {"status": "ok"}

# ============================================================
if __name__ == "__main__":
    print("=" * 50)
    print("🔑 License Server Starting...")
    print(f"   Port: {PORT}")
    print(f"   Licenses file: {LICENSE_FILE.absolute()}")
    print("=" * 50)
    uvicorn.run(app, host="0.0.0.0", port=PORT)