Creare un sistema di gestione delle licenze software con Python
Un sistema di licenze software serve a rispondere a tre domande: chi è il cliente, che cosa ha acquistato e per quanto tempo può usarlo. Tutto il resto (attivazioni, posti disponibili, revoche, periodi di grazia) è infrastruttura costruita attorno a queste tre risposte.
In questo articolo costruiremo un sistema completo in Python basato su firme digitali asimmetriche Ed25519, con licenze autocontenute verificabili offline, un server di attivazione per il conteggio dei posti e un meccanismo di revoca. Il codice è pensato per essere usato in produzione, non come giocattolo didattico.
Modello di minaccia e obiettivi realistici
Prima di scrivere una riga di codice conviene essere onesti su cosa un sistema di licenze può e non può fare. Se il software gira sulla macchina dell'utente, l'utente ha il controllo completo del processo: può patchare il bytecode, intercettare le chiamate di rete, sostituire la chiave pubblica incorporata nel binario. Nessuna tecnica lo impedisce in modo definitivo.
Gli obiettivi realistici sono quindi:
- Impedire la creazione di licenze valide da parte di terzi. Questo è un obiettivo crittograficamente raggiungibile: senza la chiave privata nessuno può forgiare una licenza.
- Impedire la modifica di una licenza esistente. Cambiare la data di scadenza o il numero di posti deve invalidare la firma.
- Rendere il conteggio dei posti verificabile. Una licenza da 5 postazioni non deve poter essere attivata su 500 macchine senza che l'azienda se ne accorga.
- Alzare il costo dell'aggiramento al punto in cui, per un cliente aziendale, pagare la licenza costa meno che violarla.
Quello che non è un obiettivo: fermare un attaccante determinato con un disassemblatore. Progettare il sistema come se lo fosse porta a complessità inutile e a bug che colpiscono i clienti paganti.
Architettura del sistema
Il sistema si articola su tre componenti separati:
- L'issuer (emittente): un tool a riga di comando che gira su una macchina controllata dal vendor e custodisce la chiave privata. Produce token di licenza firmati.
- Il server di attivazione: un servizio HTTP che registra le attivazioni, applica il limite dei posti, emette lease a breve scadenza e pubblica la lista di revoca.
- Il client: una libreria incorporata nell'applicazione che verifica la licenza offline con la chiave pubblica, gestisce la cache del lease e applica il periodo di grazia.
La scelta chiave è che la licenza è autocontenuta e verificabile offline. Il server serve solo per il conteggio dei posti e per la revoca: se il server è irraggiungibile, l'applicazione continua a funzionare fino alla scadenza del lease più il periodo di grazia. Questo elimina la peggiore modalità di guasto di un sistema di licenze, ovvero il blocco di tutti i clienti quando il servizio del vendor va offline.
Useremo Ed25519 invece di RSA per tre motivi: le chiavi pubbliche sono di 32 byte (facili da incorporare in una costante nel sorgente), le firme sono di 64 byte (il token resta corto), e l'implementazione non ha i parametri configurabili di RSA che si possono sbagliare.
Preparazione dell'ambiente
python -m venv .venv
source .venv/bin/activate
pip install cryptography fastapi "uvicorn[standard]" pydantic httpx pytestLa struttura del progetto:
licensing/
├── __init__.py
├── model.py # dataclass della licenza e serializzazione canonica
├── keys.py # generazione e caricamento delle chiavi
├── issuer.py # firma dei token
├── verifier.py # verifica offline
├── fingerprint.py # identificazione della macchina
├── lease.py # token di attivazione a breve scadenza
├── gate.py # integrazione runtime nell'applicazione
└── cli.py # tool di emissione
server/
├── __init__.py
├── db.py
└── app.py
tests/
└── test_licensing.pyIl modello dati della licenza
La licenza è una struttura immutabile. Ogni campo che finisce nella firma deve essere esplicito: quello che non viene firmato non è affidabile.
# licensing/model.py
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import datetime, timezone
LICENSE_SCHEMA_VERSION = 1
def to_iso(value: datetime) -> str:
# Rifiutiamo i datetime naive: un fuso orario implicito e' una fonte
# inesauribile di bug nelle scadenze
if value.tzinfo is None:
raise ValueError("naive datetime is not allowed")
normalized = value.astimezone(timezone.utc).replace(microsecond=0)
return normalized.isoformat().replace("+00:00", "Z")
def from_iso(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
@dataclass(frozen=True, slots=True)
class License:
license_id: str
product_id: str
customer_id: str
edition: str
seats: int
issued_at: datetime
expires_at: datetime | None = None
features: tuple[str, ...] = ()
schema_version: int = LICENSE_SCHEMA_VERSION
def to_payload(self) -> dict:
# Chiavi corte: il payload finisce in base64 dentro una stringa
# che il cliente dovra' copiare e incollare
return {
"v": self.schema_version,
"lid": self.license_id,
"pid": self.product_id,
"cid": self.customer_id,
"ed": self.edition,
"seats": self.seats,
"iat": to_iso(self.issued_at),
"exp": to_iso(self.expires_at) if self.expires_at else None,
"feat": sorted(self.features),
}
@classmethod
def from_payload(cls, payload: dict) -> "License":
expires_raw = payload.get("exp")
return cls(
license_id=payload["lid"],
product_id=payload["pid"],
customer_id=payload["cid"],
edition=payload["ed"],
seats=int(payload["seats"]),
issued_at=from_iso(payload["iat"]),
expires_at=from_iso(expires_raw) if expires_raw else None,
features=tuple(payload.get("feat", ())),
schema_version=int(payload["v"]),
)
def has_feature(self, name: str) -> bool:
return name in self.features
@property
def is_perpetual(self) -> bool:
return self.expires_at is None
def canonical_bytes(payload: dict) -> bytes:
# Serializzazione canonica: chiavi ordinate e separatori compatti.
# Lo stesso dizionario deve produrre sempre gli stessi byte, altrimenti
# la firma calcolata dall'issuer non corrisponde a quella verificata
# dal client.
return json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")Il punto delicato è canonical_bytes. Un errore classico consiste nel firmare il payload deserializzato e riserializzato con impostazioni diverse: basta uno spazio dopo i due punti perché la verifica fallisca su una licenza perfettamente valida. Nella verifica, come vedremo, firmeremo e verificheremo sempre i byte trasmessi, mai una loro ricostruzione.
Generazione e gestione delle chiavi
# licensing/keys.py
from __future__ import annotations
import base64
import hashlib
from pathlib import Path
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
def generate_keypair(
private_path: Path,
public_path: Path,
password: bytes | None = None,
) -> tuple[Ed25519PrivateKey, Ed25519PublicKey]:
private_key = Ed25519PrivateKey.generate()
public_key = private_key.public_key()
encryption = (
serialization.BestAvailableEncryption(password)
if password
else serialization.NoEncryption()
)
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=encryption,
)
public_pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
private_path.write_bytes(private_pem)
# La chiave privata non deve essere leggibile da altri utenti del sistema
private_path.chmod(0o600)
public_path.write_bytes(public_pem)
return private_key, public_key
def load_private_key(path: Path, password: bytes | None = None) -> Ed25519PrivateKey:
key = serialization.load_pem_private_key(path.read_bytes(), password=password)
if not isinstance(key, Ed25519PrivateKey):
raise TypeError("expected an Ed25519 private key")
return key
def load_public_key(path: Path) -> Ed25519PublicKey:
key = serialization.load_pem_public_key(path.read_bytes())
if not isinstance(key, Ed25519PublicKey):
raise TypeError("expected an Ed25519 public key")
return key
def public_key_raw(public_key: Ed25519PublicKey) -> bytes:
# 32 byte grezzi: la forma piu' compatta per incorporare la chiave
# direttamente nel sorgente dell'applicazione
return public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
def public_key_from_raw(data: bytes) -> Ed25519PublicKey:
return Ed25519PublicKey.from_public_bytes(data)
def key_id(public_key: Ed25519PublicKey) -> str:
# Identificatore breve e stabile della chiave: serve per la rotazione,
# perche' il client deve sapere quale chiave usare per la verifica
digest = hashlib.sha256(public_key_raw(public_key)).digest()
return base64.urlsafe_b64encode(digest[:6]).decode("ascii").rstrip("=")Il key_id è il dettaglio che rende possibile la rotazione delle chiavi senza invalidare le licenze già emesse. Il client mantiene un dizionario di chiavi pubbliche indicizzate per kid: quando si introduce una nuova chiave, si aggiunge al dizionario nella versione successiva dell'applicazione e le licenze vecchie continuano a verificarsi.
L'issuer: firma dei token
Il formato del token è composto da tre parti separate da punto: un prefisso di versione, il payload in base64url e la firma in base64url.
# licensing/issuer.py
from __future__ import annotations
import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from .keys import key_id
from .model import License, canonical_bytes
TOKEN_PREFIX = "LIC1"
def b64encode(raw: bytes) -> str:
# base64url senza padding: il token resta copiabile senza escaping
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def b64decode(value: str) -> bytes:
padding = "=" * (-len(value) % 4)
return base64.urlsafe_b64decode(value + padding)
class LicenseIssuer:
def __init__(self, private_key: Ed25519PrivateKey) -> None:
self._private_key = private_key
self._key_id = key_id(private_key.public_key())
def issue(self, license_data: License) -> str:
payload = license_data.to_payload()
# Il kid viene inserito nel payload firmato: se un attaccante lo
# modificasse per puntare a un'altra chiave, la firma non tornerebbe
payload["kid"] = self._key_id
message = canonical_bytes(payload)
signature = self._private_key.sign(message)
return f"{TOKEN_PREFIX}.{b64encode(message)}.{b64encode(signature)}"
def format_for_humans(token: str, group_size: int = 48) -> str:
# Spezza il token in righe leggibili per email e file di licenza
return "\n".join(
token[index : index + group_size]
for index in range(0, len(token), group_size)
)Un token tipico ha una lunghezza di circa 300 caratteri. Non è una chiave da cinque gruppi di cinque lettere che l'utente digita a mano: è un blocco di testo da incollare o un file license.key da posizionare nella directory di configurazione. Se serve una chiave breve digitabile, il modello corretto è diverso — una stringa casuale opaca che il server risolve in una licenza vera — e lo tratteremo nella sezione sull'attivazione.
Il verifier: validazione offline
La verifica è il componente più importante del sistema, perché gira sulla macchina del cliente e determina se l'applicazione parte. Deve essere rigorosa e produrre errori distinguibili: un utente con una licenza scaduta merita un messaggio diverso da uno con una licenza corrotta.
# licensing/verifier.py
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from .issuer import TOKEN_PREFIX, b64decode
from .model import License
class LicenseError(Exception):
"""Base class for every licensing failure."""
class MalformedLicense(LicenseError):
pass
class BadSignature(LicenseError):
pass
class UnknownSigningKey(LicenseError):
pass
class ProductMismatch(LicenseError):
pass
class LicenseExpired(LicenseError):
def __init__(self, expired_at: datetime) -> None:
super().__init__(f"license expired at {expired_at.isoformat()}")
self.expired_at = expired_at
class LicenseVerifier:
def __init__(
self,
public_keys: dict[str, Ed25519PublicKey],
product_id: str,
clock_skew: timedelta = timedelta(hours=24),
) -> None:
self._public_keys = public_keys
self._product_id = product_id
self._clock_skew = clock_skew
def verify(self, token: str, now: datetime | None = None) -> License:
now = now or datetime.now(timezone.utc)
payload_dict, message = self._decode_and_authenticate(token)
license_data = License.from_payload(payload_dict)
if license_data.product_id != self._product_id:
raise ProductMismatch(
f"license is for product {license_data.product_id!r}"
)
if license_data.expires_at is not None:
# La tolleranza sull'orologio evita falsi negativi su macchine
# con l'ora leggermente sfasata
if now > license_data.expires_at + self._clock_skew:
raise LicenseExpired(license_data.expires_at)
return license_data
def _decode_and_authenticate(self, token: str) -> tuple[dict, bytes]:
cleaned = "".join(token.split())
parts = cleaned.split(".")
if len(parts) != 3 or parts[0] != TOKEN_PREFIX:
raise MalformedLicense("unrecognized license format")
try:
message = b64decode(parts[1])
signature = b64decode(parts[2])
except (ValueError, TypeError) as exc:
raise MalformedLicense("license is not valid base64") from exc
try:
payload_dict = json.loads(message.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise MalformedLicense("license payload is not valid JSON") from exc
if not isinstance(payload_dict, dict):
raise MalformedLicense("license payload must be an object")
signing_key_id = payload_dict.get("kid")
public_key = self._public_keys.get(signing_key_id)
if public_key is None:
raise UnknownSigningKey(f"unknown signing key {signing_key_id!r}")
try:
# Verifichiamo i byte trasmessi, non una loro riserializzazione:
# e' l'unico modo per essere immuni alle differenze di formato
public_key.verify(signature, message)
except InvalidSignature as exc:
raise BadSignature("license signature does not match") from exc
return payload_dict, messageDa notare l'ordine delle operazioni: la firma viene verificata prima di interpretare qualunque campo del payload come dato affidabile. L'unico campo letto prima della verifica è kid, che serve a scegliere la chiave; ma poiché kid è a sua volta dentro il messaggio firmato, un attaccante non può alterarlo senza rompere la firma.
La chiave pubblica viene incorporata nel sorgente dell'applicazione:
# myapp/licensing_config.py
import base64
from licensing.keys import public_key_from_raw
# Chiave pubblica di produzione, 32 byte in base64.
# Sostituirla richiede una nuova release: e' esattamente cio' che vogliamo
TRUSTED_KEYS_RAW = {
"Kx7dQmVs": "TDoTvKAyoY0tOmpfy2Ky1sHNXjHFTs6MOXXPh0MKGYE=",
}
TRUSTED_KEYS = {
identifier: public_key_from_raw(base64.b64decode(encoded))
for identifier, encoded in TRUSTED_KEYS_RAW.items()
}
PRODUCT_ID = "com.example.analytics"Identificazione della macchina
Per legare un'attivazione a una postazione serve un identificatore stabile. Ogni sistema operativo ne espone uno: /etc/machine-id su Linux, MachineGuid nel registro su Windows, IOPlatformUUID su macOS. Nessuno di questi va inviato in chiaro al server: si trasmette un HMAC, così il server non accumula identificatori hardware reali dei clienti.
# licensing/fingerprint.py
from __future__ import annotations
import hashlib
import hmac
import platform
import re
import subprocess
import uuid
from pathlib import Path
# Il salt lega il fingerprint al prodotto: lo stesso computer produce
# identificatori diversi per applicazioni diverse
FINGERPRINT_SALT = b"com.example.analytics/fingerprint/v1"
def _read_first_existing(paths: tuple[str, ...]) -> str | None:
for raw_path in paths:
path = Path(raw_path)
if path.is_file():
content = path.read_text(encoding="utf-8", errors="ignore").strip()
if content:
return content
return None
def _windows_machine_guid() -> str | None:
try:
import winreg
except ImportError:
return None
try:
access = winreg.KEY_READ | winreg.KEY_WOW64_64KEY
with winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Microsoft\Cryptography",
0,
access,
) as handle:
value, _ = winreg.QueryValueEx(handle, "MachineGuid")
return str(value)
except OSError:
return None
def _macos_platform_uuid() -> str | None:
try:
completed = subprocess.run(
["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"],
capture_output=True,
text=True,
timeout=5,
check=True,
)
except (OSError, subprocess.SubprocessError):
return None
match = re.search(r'"IOPlatformUUID"\s*=\s*"([^"]+)"', completed.stdout)
return match.group(1) if match else None
def raw_machine_id() -> str:
system = platform.system()
if system == "Linux":
value = _read_first_existing(
("/etc/machine-id", "/var/lib/dbus/machine-id")
)
elif system == "Windows":
value = _windows_machine_guid()
elif system == "Darwin":
value = _macos_platform_uuid()
else:
value = None
# Ripiego sull'indirizzo MAC: meno stabile, ma sempre disponibile
return value or f"mac-{uuid.getnode():012x}"
def machine_fingerprint() -> str:
digest = hmac.new(
FINGERPRINT_SALT,
raw_machine_id().encode("utf-8"),
hashlib.sha256,
)
return digest.hexdigest()[:32]Attenzione a due trappole. La prima: nei container Docker /etc/machine-id è spesso identico su tutte le istanze costruite dalla stessa immagine, oppure cambia a ogni riavvio. Se il prodotto è destinato a girare in container, occorre prevedere una variabile d'ambiente che permetta di fissare il fingerprint esplicitamente. La seconda: non combinare troppi segnali hardware (CPU, dischi, schede di rete) sperando in maggiore robustezza, perché il risultato è che il fingerprint cambia quando l'utente aggiunge un disco e la licenza si "consuma" un posto senza motivo. Un solo identificatore stabile, con un ripiego, è la scelta corretta.
Il lease: token di attivazione a breve scadenza
Il server non risponde con un semplice "ok". Restituisce un lease: un token firmato, valido pochi giorni, che lega la licenza a una macchina specifica. Il client lo memorizza in cache e lo verifica offline a ogni avvio. Questo è ciò che rende il conteggio dei posti applicabile senza richiedere connettività continua.
# licensing/lease.py
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
from .issuer import b64decode, b64encode
from .keys import key_id
from .model import License, canonical_bytes, from_iso, to_iso
from .verifier import BadSignature, MalformedLicense, UnknownSigningKey
LEASE_PREFIX = "LEA1"
@dataclass(frozen=True, slots=True)
class Lease:
license_id: str
product_id: str
machine_hash: str
issued_at: datetime
expires_at: datetime
seats: int
features: tuple[str, ...]
def is_valid_at(self, moment: datetime) -> bool:
return moment <= self.expires_at
def issue_lease(
private_key: Ed25519PrivateKey,
license_data: License,
machine_hash: str,
ttl: timedelta,
) -> str:
now = datetime.now(timezone.utc)
payload = {
"kid": key_id(private_key.public_key()),
"lid": license_data.license_id,
"pid": license_data.product_id,
"mh": machine_hash,
"iat": to_iso(now),
"exp": to_iso(now + ttl),
"seats": license_data.seats,
"feat": sorted(license_data.features),
}
message = canonical_bytes(payload)
signature = private_key.sign(message)
return f"{LEASE_PREFIX}.{b64encode(message)}.{b64encode(signature)}"
def verify_lease(
token: str,
public_keys: dict[str, Ed25519PublicKey],
expected_machine_hash: str,
) -> Lease:
parts = "".join(token.split()).split(".")
if len(parts) != 3 or parts[0] != LEASE_PREFIX:
raise MalformedLicense("unrecognized lease format")
message = b64decode(parts[1])
signature = b64decode(parts[2])
payload = json.loads(message.decode("utf-8"))
public_key = public_keys.get(payload.get("kid"))
if public_key is None:
raise UnknownSigningKey("unknown lease signing key")
try:
public_key.verify(signature, message)
except InvalidSignature as exc:
raise BadSignature("lease signature does not match") from exc
# Un lease copiato su un'altra macchina non deve funzionare
if payload["mh"] != expected_machine_hash:
raise BadSignature("lease was issued for a different machine")
return Lease(
license_id=payload["lid"],
product_id=payload["pid"],
machine_hash=payload["mh"],
issued_at=from_iso(payload["iat"]),
expires_at=from_iso(payload["exp"]),
seats=int(payload["seats"]),
features=tuple(payload.get("feat", ())),
)Il server di attivazione
Il server ha una responsabilità circoscritta: sapere quante macchine hanno attivato ciascuna licenza e rifiutare le attivazioni oltre il limite. Non ha bisogno di conoscere le licenze in anticipo, perché il token è autocontenuto e firmato: la riga nel database viene creata alla prima attivazione.
Lo schema del database
# server/db.py
from __future__ import annotations
import sqlite3
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator
DB_PATH = Path("licensing.db")
SCHEMA = """
CREATE TABLE IF NOT EXISTS licenses (
license_id TEXT PRIMARY KEY,
product_id TEXT NOT NULL,
customer_id TEXT NOT NULL,
seats INTEGER NOT NULL,
first_seen_at TEXT NOT NULL,
revoked_at TEXT,
revoke_reason TEXT
);
CREATE TABLE IF NOT EXISTS activations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
license_id TEXT NOT NULL REFERENCES licenses(license_id),
machine_hash TEXT NOT NULL,
hostname TEXT,
activated_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL,
released_at TEXT,
UNIQUE (license_id, machine_hash)
);
CREATE INDEX IF NOT EXISTS idx_activations_active
ON activations (license_id) WHERE released_at IS NULL;
"""
@contextmanager
def connection() -> Iterator[sqlite3.Connection]:
# isolation_level=None disattiva la gestione implicita delle transazioni:
# le controlliamo esplicitamente con BEGIN IMMEDIATE
conn = sqlite3.connect(DB_PATH, isolation_level=None, timeout=10.0)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
finally:
conn.close()
def init_db() -> None:
with connection() as conn:
conn.executescript(SCHEMA)Il vincolo UNIQUE (license_id, machine_hash) è la garanzia strutturale contro il doppio conteggio: una macchina che si riattiva aggiorna la riga esistente invece di crearne una nuova. La colonna released_at permette il rilascio del posto (disattivazione) senza cancellare lo storico.
Gli endpoint
# server/app.py
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from licensing.keys import key_id, load_private_key, load_public_key
from licensing.lease import issue_lease
from licensing.model import to_iso
from licensing.verifier import LicenseError, LicenseVerifier
from .db import connection, init_db
PRODUCT_ID = "com.example.analytics"
LEASE_TTL = timedelta(days=7)
private_key = load_private_key(Path("/etc/licensing/private.pem"))
public_key = load_public_key(Path("/etc/licensing/public.pem"))
verifier = LicenseVerifier({key_id(public_key): public_key}, PRODUCT_ID)
app = FastAPI(title="License Activation Service")
@app.on_event("startup")
def on_startup() -> None:
init_db()
class ActivationRequest(BaseModel):
license_token: str = Field(min_length=32, max_length=8192)
machine_hash: str = Field(min_length=16, max_length=64, pattern=r"^[0-9a-f]+$")
hostname: str | None = Field(default=None, max_length=255)
class ActivationResponse(BaseModel):
lease_token: str
seats_used: int
seats_total: int
expires_at: str
def now_iso() -> str:
return to_iso(datetime.now(timezone.utc))
@app.post("/v1/activate", response_model=ActivationResponse)
def activate(request: ActivationRequest) -> ActivationResponse:
try:
license_data = verifier.verify(request.license_token)
except LicenseError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
with connection() as conn:
# BEGIN IMMEDIATE acquisisce subito il lock di scrittura: senza,
# due attivazioni concorrenti potrebbero superare il limite di posti
conn.execute("BEGIN IMMEDIATE")
try:
row = conn.execute(
"SELECT revoked_at FROM licenses WHERE license_id = ?",
(license_data.license_id,),
).fetchone()
if row is None:
conn.execute(
"INSERT INTO licenses "
"(license_id, product_id, customer_id, seats, first_seen_at) "
"VALUES (?, ?, ?, ?, ?)",
(
license_data.license_id,
license_data.product_id,
license_data.customer_id,
license_data.seats,
now_iso(),
),
)
elif row["revoked_at"]:
raise HTTPException(status_code=403, detail="license revoked")
existing = conn.execute(
"SELECT id FROM activations "
"WHERE license_id = ? AND machine_hash = ?",
(license_data.license_id, request.machine_hash),
).fetchone()
if existing is None:
used = conn.execute(
"SELECT COUNT(*) AS total FROM activations "
"WHERE license_id = ? AND released_at IS NULL",
(license_data.license_id,),
).fetchone()["total"]
if used >= license_data.seats:
raise HTTPException(
status_code=409,
detail=f"all {license_data.seats} seats are in use",
)
conn.execute(
"INSERT INTO activations "
"(license_id, machine_hash, hostname, activated_at, last_seen_at) "
"VALUES (?, ?, ?, ?, ?)",
(
license_data.license_id,
request.machine_hash,
request.hostname,
now_iso(),
now_iso(),
),
)
else:
# Riattivazione della stessa macchina: non consuma un posto
conn.execute(
"UPDATE activations "
"SET released_at = NULL, last_seen_at = ?, hostname = ? "
"WHERE id = ?",
(now_iso(), request.hostname, existing["id"]),
)
seats_used = conn.execute(
"SELECT COUNT(*) AS total FROM activations "
"WHERE license_id = ? AND released_at IS NULL",
(license_data.license_id,),
).fetchone()["total"]
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
lease_token = issue_lease(
private_key, license_data, request.machine_hash, LEASE_TTL
)
return ActivationResponse(
lease_token=lease_token,
seats_used=seats_used,
seats_total=license_data.seats,
expires_at=to_iso(datetime.now(timezone.utc) + LEASE_TTL),
)
class ReleaseRequest(BaseModel):
license_token: str
machine_hash: str = Field(pattern=r"^[0-9a-f]+$")
@app.post("/v1/release", status_code=204)
def release(request: ReleaseRequest) -> None:
try:
license_data = verifier.verify(request.license_token)
except LicenseError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
with connection() as conn:
conn.execute(
"UPDATE activations SET released_at = ? "
"WHERE license_id = ? AND machine_hash = ? AND released_at IS NULL",
(now_iso(), license_data.license_id, request.machine_hash),
)La chiamata a BEGIN IMMEDIATE merita attenzione. Con SQLite in modalità di default, la sequenza "conta i posti usati, poi inserisci" è soggetta a una race condition: due richieste simultanee leggono entrambe 4 posti su 5 e inseriscono entrambe, arrivando a 6. Acquisire il lock di scrittura all'inizio della transazione serializza le attivazioni. Su PostgreSQL l'equivalente è SELECT ... FOR UPDATE sulla riga della licenza.
Revoca delle licenze
Una licenza può dover essere invalidata prima della scadenza: rimborso, chargeback, violazione contrattuale. Poiché il token è autocontenuto e verificabile offline, la revoca non può essere immediata: si propaga alla scadenza del lease. È un compromesso accettabile con lease di durata breve.
Il server pubblica una lista di revoca firmata:
# server/app.py (continua)
class RevocationList(BaseModel):
issued_at: str
sequence: int
revoked_ids: list[str]
signature: str
@app.get("/v1/revocations", response_model=RevocationList)
def revocations() -> RevocationList:
from licensing.issuer import b64encode
from licensing.model import canonical_bytes
with connection() as conn:
rows = conn.execute(
"SELECT license_id FROM licenses WHERE revoked_at IS NOT NULL "
"ORDER BY license_id"
).fetchall()
payload = {
"iat": now_iso(),
# Il numero di sequenza monotono impedisce a un attaccante di
# rimettere in circolo una lista vecchia in cui la licenza revocata
# non era ancora presente
"seq": int(datetime.now(timezone.utc).timestamp()),
"ids": [row["license_id"] for row in rows],
}
signature = private_key.sign(canonical_bytes(payload))
return RevocationList(
issued_at=payload["iat"],
sequence=payload["seq"],
revoked_ids=payload["ids"],
signature=b64encode(signature),
)La firma sulla lista è indispensabile: senza, chiunque controlli la rete potrebbe consegnare al client una lista vuota, oppure una lista che revoca le licenze dei concorrenti. Il numero di sequenza monotono protegge dagli attacchi di replay: il client rifiuta una lista con sequenza inferiore all'ultima memorizzata.
Il gate di licenza nell'applicazione
Ecco il componente che l'applicazione chiama all'avvio. Incapsula tutta la logica di cache, rinnovo e periodo di grazia.
# licensing/gate.py
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
import httpx
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from .fingerprint import machine_fingerprint
from .lease import Lease, verify_lease
from .model import License, from_iso, to_iso
from .verifier import LicenseError, LicenseVerifier
@dataclass(frozen=True, slots=True)
class Entitlement:
license_data: License
lease: Lease | None
offline: bool
grace_days_left: int | None
class ActivationRequired(LicenseError):
pass
class GracePeriodExpired(LicenseError):
pass
class ClockTampering(LicenseError):
pass
class LicenseGate:
def __init__(
self,
license_path: Path,
state_dir: Path,
verifier: LicenseVerifier,
public_keys: dict[str, Ed25519PublicKey],
api_url: str,
grace_period: timedelta = timedelta(days=14),
) -> None:
self._license_path = license_path
self._state_dir = state_dir
self._verifier = verifier
self._public_keys = public_keys
self._api_url = api_url.rstrip("/")
self._grace_period = grace_period
self._lease_path = state_dir / "lease.token"
self._clock_path = state_dir / "clock.json"
def check(self) -> Entitlement:
now = self._checked_now()
license_data = self._load_license(now)
machine_hash = machine_fingerprint()
lease = self._load_cached_lease(machine_hash)
if lease is not None and lease.is_valid_at(now):
return Entitlement(license_data, lease, offline=False, grace_days_left=None)
# Lease assente o scaduto: proviamo il rinnovo online
try:
lease = self._activate_online(license_data, machine_hash)
except (httpx.HTTPError, OSError):
return self._fall_back_to_grace(license_data, lease, now)
return Entitlement(license_data, lease, offline=False, grace_days_left=None)
def _load_license(self, now: datetime) -> License:
if not self._license_path.is_file():
raise ActivationRequired("no license file found")
token = self._license_path.read_text(encoding="utf-8")
return self._verifier.verify(token, now=now)
def _load_cached_lease(self, machine_hash: str) -> Lease | None:
if not self._lease_path.is_file():
return None
try:
return verify_lease(
self._lease_path.read_text(encoding="utf-8"),
self._public_keys,
machine_hash,
)
except LicenseError:
# Un lease corrotto o appartenente ad altra macchina viene
# semplicemente ignorato: si ripiega sull'attivazione online
return None
def _activate_online(self, license_data: License, machine_hash: str) -> Lease:
import socket
response = httpx.post(
f"{self._api_url}/v1/activate",
json={
"license_token": self._license_path.read_text(encoding="utf-8"),
"machine_hash": machine_hash,
"hostname": socket.gethostname()[:255],
},
timeout=10.0,
)
if response.status_code == 409:
raise ActivationRequired("no seats available for this license")
if response.status_code == 403:
raise ActivationRequired("this license has been revoked")
response.raise_for_status()
lease_token = response.json()["lease_token"]
self._state_dir.mkdir(parents=True, exist_ok=True)
self._lease_path.write_text(lease_token, encoding="utf-8")
return verify_lease(lease_token, self._public_keys, machine_hash)
def _fall_back_to_grace(
self, license_data: License, lease: Lease | None, now: datetime
) -> Entitlement:
if lease is None:
raise ActivationRequired("first activation requires connectivity")
deadline = lease.expires_at + self._grace_period
if now > deadline:
raise GracePeriodExpired("offline grace period is over")
days_left = max(0, (deadline - now).days)
return Entitlement(license_data, lease, offline=True, grace_days_left=days_left)
def _checked_now(self) -> datetime:
# Difesa contro lo spostamento indietro dell'orologio di sistema:
# memorizziamo il timestamp massimo osservato finora
now = datetime.now(timezone.utc)
if self._clock_path.is_file():
try:
stored = from_iso(json.loads(self._clock_path.read_text())["max_seen"])
except (ValueError, KeyError, json.JSONDecodeError):
stored = None
if stored is not None and now < stored - timedelta(hours=48):
raise ClockTampering("system clock moved backwards")
self._state_dir.mkdir(parents=True, exist_ok=True)
self._clock_path.write_text(json.dumps({"max_seen": to_iso(now)}))
return nowL'uso nell'applicazione resta banale:
# myapp/main.py
import sys
from datetime import timedelta
from pathlib import Path
from licensing.gate import (
ActivationRequired,
GracePeriodExpired,
LicenseGate,
)
from licensing.verifier import LicenseError, LicenseExpired, LicenseVerifier
from .licensing_config import PRODUCT_ID, TRUSTED_KEYS
def start_application() -> int:
gate = LicenseGate(
license_path=Path.home() / ".config" / "analytics" / "license.key",
state_dir=Path.home() / ".local" / "state" / "analytics",
verifier=LicenseVerifier(TRUSTED_KEYS, PRODUCT_ID),
public_keys=TRUSTED_KEYS,
api_url="https://licensing.example.com",
grace_period=timedelta(days=14),
)
try:
entitlement = gate.check()
except LicenseExpired as exc:
print(f"License expired on {exc.expired_at:%d/%m/%Y}.", file=sys.stderr)
return 2
except ActivationRequired as exc:
print(f"Activation required: {exc}", file=sys.stderr)
return 3
except GracePeriodExpired:
print("Offline grace period is over. Connect to the internet.", file=sys.stderr)
return 4
except LicenseError as exc:
print(f"License check failed: {exc}", file=sys.stderr)
return 5
if entitlement.offline and entitlement.grace_days_left is not None:
print(f"Offline mode: {entitlement.grace_days_left} days left.")
# Le funzionalita' opzionali sono attivate dal campo firmato "feat"
if entitlement.license_data.has_feature("advanced-reporting"):
enable_advanced_reporting()
return run_main_loop()Il tool di emissione
# licensing/cli.py
from __future__ import annotations
import argparse
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from .issuer import LicenseIssuer, format_for_humans
from .keys import generate_keypair, key_id, load_private_key, load_public_key
from .model import License
from .verifier import LicenseVerifier
def cmd_keygen(args: argparse.Namespace) -> int:
password = args.password.encode("utf-8") if args.password else None
_, public_key = generate_keypair(
Path(args.private_out), Path(args.public_out), password
)
print(f"key id: {key_id(public_key)}")
return 0
def cmd_issue(args: argparse.Namespace) -> int:
private_key = load_private_key(
Path(args.private_key),
args.password.encode("utf-8") if args.password else None,
)
now = datetime.now(timezone.utc)
license_data = License(
license_id=str(uuid.uuid4()),
product_id=args.product,
customer_id=args.customer,
edition=args.edition,
seats=args.seats,
issued_at=now,
expires_at=None if args.perpetual else now + timedelta(days=args.days),
features=tuple(args.feature or ()),
)
token = LicenseIssuer(private_key).issue(license_data)
output = format_for_humans(token)
if args.output:
Path(args.output).write_text(output, encoding="utf-8")
else:
print(output)
return 0
def cmd_inspect(args: argparse.Namespace) -> int:
public_key = load_public_key(Path(args.public_key))
verifier = LicenseVerifier({key_id(public_key): public_key}, args.product)
license_data = verifier.verify(Path(args.license).read_text(encoding="utf-8"))
print(f"id : {license_data.license_id}")
print(f"customer : {license_data.customer_id}")
print(f"edition : {license_data.edition}")
print(f"seats : {license_data.seats}")
print(f"expires : {license_data.expires_at or 'never'}")
print(f"features : {', '.join(license_data.features) or 'none'}")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="licensetool")
subparsers = parser.add_subparsers(required=True)
keygen = subparsers.add_parser("keygen")
keygen.add_argument("--private-out", default="private.pem")
keygen.add_argument("--public-out", default="public.pem")
keygen.add_argument("--password")
keygen.set_defaults(func=cmd_keygen)
issue = subparsers.add_parser("issue")
issue.add_argument("--private-key", required=True)
issue.add_argument("--password")
issue.add_argument("--product", required=True)
issue.add_argument("--customer", required=True)
issue.add_argument("--edition", default="standard")
issue.add_argument("--seats", type=int, default=1)
issue.add_argument("--days", type=int, default=365)
issue.add_argument("--perpetual", action="store_true")
issue.add_argument("--feature", action="append")
issue.add_argument("--output")
issue.set_defaults(func=cmd_issue)
inspect = subparsers.add_parser("inspect")
inspect.add_argument("--public-key", required=True)
inspect.add_argument("--product", required=True)
inspect.add_argument("--license", required=True)
inspect.set_defaults(func=cmd_inspect)
return parser
def main() -> int:
args = build_parser().parse_args()
return args.func(args)L'emissione di una licenza diventa quindi:
python -m licensing.cli keygen --private-out private.pem --public-out public.pem
python -m licensing.cli issue \
--private-key private.pem \
--product com.example.analytics \
--customer acme-corp \
--edition enterprise \
--seats 25 \
--days 365 \
--feature advanced-reporting \
--feature sso \
--output acme.keyTest
I test di un sistema di licenze devono coprire soprattutto i casi negativi: è lì che si nascondono le vulnerabilità.
# tests/test_licensing.py
from datetime import datetime, timedelta, timezone
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from licensing.issuer import LicenseIssuer, b64decode, b64encode
from licensing.keys import key_id
from licensing.model import License
from licensing.verifier import (
BadSignature,
LicenseExpired,
LicenseVerifier,
MalformedLicense,
ProductMismatch,
UnknownSigningKey,
)
PRODUCT = "com.example.analytics"
@pytest.fixture
def keypair():
private_key = Ed25519PrivateKey.generate()
return private_key, private_key.public_key()
@pytest.fixture
def verifier(keypair):
_, public_key = keypair
return LicenseVerifier({key_id(public_key): public_key}, PRODUCT)
def make_license(**overrides) -> License:
now = datetime.now(timezone.utc)
defaults = {
"license_id": "11111111-2222-3333-4444-555555555555",
"product_id": PRODUCT,
"customer_id": "acme-corp",
"edition": "enterprise",
"seats": 25,
"issued_at": now,
"expires_at": now + timedelta(days=365),
"features": ("sso",),
}
defaults.update(overrides)
return License(**defaults)
def test_valid_license_round_trip(keypair, verifier):
private_key, _ = keypair
token = LicenseIssuer(private_key).issue(make_license())
result = verifier.verify(token)
assert result.customer_id == "acme-corp"
assert result.has_feature("sso")
def test_tampered_payload_is_rejected(keypair, verifier):
private_key, _ = keypair
token = LicenseIssuer(private_key).issue(make_license(seats=5))
prefix, payload_b64, signature_b64 = token.split(".")
# Proviamo a portare i posti da 5 a 500 lasciando la firma originale
forged = b64decode(payload_b64).replace(b'"seats":5', b'"seats":500')
forged_token = f"{prefix}.{b64encode(forged)}.{signature_b64}"
with pytest.raises(BadSignature):
verifier.verify(forged_token)
def test_license_signed_by_another_key_is_rejected(verifier):
attacker_key = Ed25519PrivateKey.generate()
token = LicenseIssuer(attacker_key).issue(make_license())
with pytest.raises(UnknownSigningKey):
verifier.verify(token)
def test_expired_license_is_rejected(keypair, verifier):
private_key, _ = keypair
expired_at = datetime.now(timezone.utc) - timedelta(days=10)
token = LicenseIssuer(private_key).issue(
make_license(expires_at=expired_at)
)
with pytest.raises(LicenseExpired):
verifier.verify(token)
def test_clock_skew_tolerance(keypair, verifier):
private_key, _ = keypair
# Scaduta da un'ora: la tolleranza di 24 ore deve accettarla
expires_at = datetime.now(timezone.utc) - timedelta(hours=1)
token = LicenseIssuer(private_key).issue(make_license(expires_at=expires_at))
assert verifier.verify(token).seats == 25
def test_perpetual_license_never_expires(keypair, verifier):
private_key, _ = keypair
token = LicenseIssuer(private_key).issue(make_license(expires_at=None))
far_future = datetime.now(timezone.utc) + timedelta(days=20000)
assert verifier.verify(token, now=far_future).is_perpetual
def test_wrong_product_is_rejected(keypair, verifier):
private_key, _ = keypair
token = LicenseIssuer(private_key).issue(
make_license(product_id="com.example.other")
)
with pytest.raises(ProductMismatch):
verifier.verify(token)
@pytest.mark.parametrize(
"token",
["", "garbage", "LIC1.only-two-parts", "LIC9.aaaa.bbbb", "LIC1.!!!.???"],
)
def test_malformed_tokens_are_rejected(verifier, token):
with pytest.raises(MalformedLicense):
verifier.verify(token)Il test sulla manomissione del payload è il più importante di tutti: verifica che il meccanismo faccia esattamente ciò per cui esiste. Vale la pena aggiungerne uno analogo che sposti la data di scadenza.
Rotazione delle chiavi
Prima o poi una chiave privata va sostituita, per compromissione o per policy. La procedura, con l'architettura descritta, è indolore:
- Si genera la nuova coppia e si annota il nuovo
key_id. - Si aggiunge la nuova chiave pubblica al dizionario
TRUSTED_KEYSdell'applicazione, mantenendo la vecchia, e si rilascia la nuova versione. - Si attende che il parco installato aggiorni (mesi, tipicamente).
- Si iniziano a emettere licenze con la nuova chiave.
- Si rimuove la vecchia chiave solo quando tutte le licenze da essa firmate sono scadute.
In caso di compromissione della chiave privata l'ordine cambia: si emette immediatamente una nuova versione che rimuove la chiave compromessa e si riemettono le licenze dei clienti legittimi. È doloroso, ed è il motivo per cui la chiave privata deve stare su una macchina isolata o in un HSM, mai in un repository e mai su un server web.
Considerazioni finali e limiti
Alcune scelte pragmatiche che vale la pena esplicitare.
Non offuscare il codice sperando che basti. Strumenti come PyArmor o Nuitka alzano il costo dell'analisi, ma un attaccante può sempre sostituire la funzione check con una che restituisce sempre successo. L'offuscamento è ragionevole come deterrente marginale, dannoso se induce a trascurare la parte crittografica.
Distribuire il controllo nel codice. Un singolo if not gate.check(): sys.exit() all'avvio è un unico punto da patchare. Verificare l'Entitlement in più punti, dove serve davvero (all'attivazione di una funzionalità premium, nel percorso di export dei dati), rende l'aggiramento più laborioso. Non renderlo però così invasivo da diventare un problema di manutenzione.
Fallire in modo gentile. Ogni percorso di errore va progettato pensando al cliente pagante che lo incontrerà per un problema di rete o un orologio sbagliato. Messaggi chiari, codici di uscita distinti, un periodo di grazia generoso e un modo per contattare l'assistenza valgono più di qualsiasi restrizione aggiuntiva.
Conservare i dati minimi. Il server riceve un HMAC del machine id e un hostname. Non servono indirizzo IP persistente, elenco dell'hardware o telemetria d'uso: raccoglierli crea obblighi GDPR sproporzionati rispetto al beneficio.
Prevedere la disattivazione self-service. Un cliente che riformatta il computer deve poter liberare il posto da solo, dall'applicazione o da un portale. Senza questa possibilità, l'assistenza si riempie di ticket e i clienti percepiscono il sistema di licenze come un ostacolo anziché come un dettaglio amministrativo.
Il sistema descritto — licenze Ed25519 autocontenute, lease a scadenza breve, conteggio dei posti lato server e revoca firmata — copre le esigenze della grande maggioranza dei prodotti commerciali, resta comprensibile in poche centinaia di righe di Python e, soprattutto, non si trasforma in un ostacolo per chi ha pagato.