Spaces:
Sleeping
Sleeping
File size: 5,466 Bytes
1e2860e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """
============================================================
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) |