Skip to content

README.md

Latest commit

 

History

History
617 lines (466 loc) · 20.2 KB

README.md

File metadata and controls

617 lines (466 loc) · 20.2 KB
Raw

Python SDK

keyvory_client.py is a single-file Python SDK for the Keyvory license server. Drop it into your project alongside pycryptodome, requests, and pynacl.

Table of Contents

Installation

Copy keyvory_client.py from example/python/ into your project. Install the three dependencies if you haven't already:

pip install pycryptodome requests pynacl

These are already in the server's requirements.txt, so if you are running the server and the client on the same machine you are all set.

Quick Start

from keyvory_client import KeyvoryClient, LicenseStatus, TamperError, NetworkError

with KeyvoryClient(
    server_url="https://licenses.yourapp.com",
    secret_key="your-secret-key",           # same SECRET_KEY as the server .env
) as client:

    # 1. Handshake — must be called before anything else
    if not client.initialize():
        raise RuntimeError("Could not reach license server")

    # 2. Validate the user's license key
    result: LicenseStatus = client.login("KEYVORY-AAAAA-BBBBB-CCCCC-DDDDD")

    if result.is_valid:
        print(f"Welcome! Your license expires {result.expiration_date}")
        if result.has_feature("dark_mode"):
            enable_dark_mode()
    else:
        print(f"License denied: {result.status}")
        sys.exit(1)

    # 3. Keep the session alive while your app runs
    while app_is_running():
        do_work()
        client.heartbeat(license_key="KEYVORY-AAAAA-BBBBB-CCCCC-DDDDD")
        time.sleep(60)

KeyvoryClient

Constructor

KeyvoryClient(
    server_url: str,
    secret_key: str,
    api_version: str = "v1",
    max_retries: int = 3,
    timeout: int = 10,
)
Parameter Type Default Description
server_url str required Base URL of your Keyvory server, e.g. https://licenses.yourapp.com
secret_key str required The SECRET_KEY from your server .env file
api_version str "v1" API version string. Change only if you have a custom build
max_retries int 3 How many times to retry on ConnectionError or Timeout
timeout int 10 Per-request timeout in seconds

KeyvoryClient is a context manager. Use it with with to ensure the underlying requests.Session is always closed:

with KeyvoryClient(...) as client:
    ...
# session is closed here automatically

You can also manage the lifecycle manually:

client = KeyvoryClient(...)
try:
    ...
finally:
    client.close()

initialize()

client.initialize() -> bool

Performs the session bootstrap handshake with the server. This must succeed before any other SDK call.

What happens:

  1. The SDK computes a bootstrap key: SHA-256("keyvory-bootstrap:{SECRET_KEY}")
  2. Sends a POST to /api/v1/license/init with headers:
    • X-Client-ID — AES-encrypted client identifier
    • X-Bootstrap-Key — the bootstrap proof above
    • X-Signature — HMAC-SHA256 of "{client_id}|{timestamp}"
    • X-Timestamp, X-Signature-Version, X-Client-Version, X-Device-ID, X-Platform
  3. Server validates the bootstrap key, decrypts the client ID, verifies the HMAC signature, and rejects duplicates
  4. Server responds with:
    • key — AES-encrypted session key (encrypted with the client ID)
    • x25519_private — X25519 private key (encrypted with the session key)
    • server_public_key — Ed25519 verify key
    • server_x25519_public — X25519 public key
    • timestamp — server timestamp (must be within 60s of client time)
  5. The SDK decrypts and stores the session key and X25519 private key, then marks itself as initialized

Returns True on success, False on any failure (network error, bad secret, clock skew > 60s). Raises TamperError if a debugger is detected.

if not client.initialize():
    # wrong URL, wrong secret key, server down, or clock skew
    raise RuntimeError("License server unreachable")

Calling initialize() on an already-initialized client is a no-op and returns True immediately.

login()

client.login(license_key: str) -> LicenseStatus

Validates a license key and binds it to the current machine's hardware fingerprint. On first call with a given key, the server locks the key to this machine (or increments the device count up to max_devices). Subsequent calls from the same machine succeed; calls from a different machine are rejected once the slot is full.

What happens:

  1. Generates an ephemeral X25519 keypair
  2. Derives a shared secret from the ephemeral private key + the server's X25519 public key
  3. Derives AES and MAC keys from the shared secret via HKDF-SHA256
  4. Encrypts the payload (license_key + hwid) with AES-256-GCM
  5. Computes HMAC-SHA256 over the ciphertext
  6. Sends {"e": ciphertext, "m": mac, "ek": ephemeral_public_key} with auth headers
  7. Server opens the envelope, validates the license, and returns the result in a sealed envelope
  8. SDK verifies the Ed25519 signature, opens the response envelope, and constructs a LicenseStatus
Parameter Type Description
license_key str The license key to validate, e.g. "KEYVORY-XXXXX-XXXXX-XXXXX-XXXXX"

Returns a LicenseStatus object. Never raises on a denial — check result.is_valid. Raises NetworkError if the server is unreachable after all retries. Raises TamperError if anti-tamper checks fail.

result = client.login("KEYVORY-XXXXX-XXXXX-XXXXX-XXXXX")

if result.is_valid:
    print(result.license_type)       # "Enterprise"
    print(result.expiration_date)    # "2027-01-01"
    print(result.max_devices)        # 5
else:
    # result.status is one of:
    #   "denied"          — key invalid or revoked
    #   "not_initialized" — you forgot to call initialize() first
    #   "error"           — unexpected server error
    print("denied:", result.status, result.code)

get_license_info()

client.get_license_info(license_key: str) -> LicenseStatus

Fetches license metadata without modifying the HWID lock or device count. Useful for displaying "your license" information in a settings screen.

The caller must already own the license (matching client_id enforced server-side). If the key belongs to a different client, the server returns an error (no information is disclosed to non-owners).

Response field Maps to LicenseStatus
status status
type license_type
application application
expiration expiration_date
features features
days_until_expiry features["days_until_expiry"] 
info = client.get_license_info("KEYVORY-XXXXX-XXXXX-XXXXX-XXXXX")
if info.status != "error":
    print(f"Type: {info.license_type}")
    print(f"Expires: {info.expiration_date}")
    print(f"Days left: {info.features.get('days_until_expiry', '?')}")

heartbeat()

client.heartbeat(license_key: Optional[str] = None) -> bool

Sends a keepalive ping and optionally re-validates the active license. Call this on a background thread or timer while your application is running.

If license_key is provided, the server re-checks validity (expiration, revocation). If the license has been revoked since the last check, the server responds with {"status": "revoked"} and heartbeat() returns False.

If license_key is omitted, the heartbeat is session-only — useful for keeping the session alive without the overhead of a full license re-check.

Returns True if the server responds with {"status": "ok"}, False otherwise. The SDK auto-rotates the session key every 100 heartbeat calls via _maybe_rotate().

import threading, time

def keep_alive(client, key):
    while True:
        alive = client.heartbeat(license_key=key)
        if not alive:
            # license was revoked — take action
            disable_app()
            break
        time.sleep(60)

t = threading.Thread(target=keep_alive, args=(client, license_key), daemon=True)
t.start()

rotate_key()

client.rotate_key() -> bool

Requests a new AES session key from the server. The flow:

  1. Server generates a new random 256-bit session key
  2. Server encrypts it with the current session key (AES-GCM(current_key, new_key))
  3. Server increments key_version and commits to the database
  4. SDK decrypts the new key using the current session key
  5. SDK swaps in the new key and increments the local key_version
  6. SDK updates the HMAC state chain

This is called automatically every 100 heartbeats via _maybe_rotate() — you rarely need to call it directly.

if client.rotate_key():
    print("Session key rotated")
else:
    print("Rotation failed — old key still in use")

close()

client.close() -> None

Closes the underlying requests.Session. Called automatically when used as a context manager.

LicenseStatus

Returned by login() and get_license_info().

class LicenseStatus:
    status: str            # "valid", "denied", "error", "not_initialized"
    expiration_date: str   # ISO-8601 date string, e.g. "2027-01-01"
    license_type: str      # name of the license type, e.g. "Professional"
    application: str       # application name the license is scoped to
    features: dict         # {"dark_mode": True, "export_pdf": False, …}
    code: str              # short denial code for logging, e.g. "EXPIRED", "REVOKED"
    max_devices: int       # maximum number of hardware IDs this key allows

Properties:

is_valid -> bool — True when status == "valid". The only field you need to gate access.

if result.is_valid:
    unlock_premium_features()

has_feature(name: str) -> bool — returns True if features[name] is truthy.

if result.has_feature("unlimited_exports"):
    remove_export_cap()

Exceptions

from keyvory_client import KeyvoryError, TamperError, NetworkError
Exception Inherits Raised when
KeyvoryError Exception Base class for all SDK exceptions
TamperError KeyvoryError Debugger detected, timing anomaly, or session-state integrity violation
NetworkError KeyvoryError Server unreachable after all retries

Denials (wrong key, expired, wrong machine) are not exceptions — they are represented as LicenseStatus objects with is_valid == False. Only infrastructure failures and security violations raise exceptions.

try:
    if not client.initialize():
        show_error("Cannot reach license server")
        sys.exit(1)

    result = client.login(key)

    if not result.is_valid:
        show_error(f"License denied: {result.status}")
        sys.exit(1)

except TamperError:
    # SDK detected a debugger or in-memory state tampering
    show_error("Integrity check failed. Please restart the application.")
    sys.exit(1)

except NetworkError as e:
    show_error(f"Network error: {e}")
    sys.exit(1)

Encryption Protocol

The SDK implements a two-layer envelope encryption scheme matching the server.

Init Handshake

Client                              Server
  │                                    │
  │  POST /api/v1/license/init         │
  │  Headers:                          │
  │    X-Client-ID    (AES-encrypted)  │
  │    X-Bootstrap-Key SHA256(...SK)   │
  │    X-Signature     HMAC(client_id|ts)
  │    X-Timestamp                     │
  │  ─────────────────────────────────> │
  │                                    │
  │  Validate bootstrap key            │
  │  Decrypt client_id                 │
  │  Verify HMAC signature             │
  │  Check for duplicate client_id     │
  │  Generate client_key + X25519 pair │
  │                                    │
  │  Response JSON:                    │
  │    key              (AES-encrypted client_key)
  │    x25519_private   (AES-encrypted with client_key)
  │    server_public_key (Ed25519)
  │    server_x25519_public            │
  │    timestamp                       │
  │  <───────────────────────────────── │
  │                                    │
  │  Decrypt client_key with client_id │
  │  Decrypt x25519_private with client_key
  │  Store both keys                   │

Sealed Envelope (login, heartbeat, info, rotate-key)

Client                              Server
  │                                    │
  │  Generate ephemeral X25519 pair    │
  │  shared_secret = X25519(eph_priv, server_pub)
  │  aes_key, mac_key = HKDF(shared_secret)
  │                                    │
  │  ciphertext = AES-GCM(aes_key, payload)
  │  mac = HMAC-SHA256(mac_key, ciphertext)
  │                                    │
  │  POST with envelope:               │
  │    {"e": ciphertext,               │
  │     "m": mac,                      │
  │     "ek": ephemeral_pub}           │
  │  + auth headers (X-Client-ID, etc) │
  │  ─────────────────────────────────> │
  │                                    │
  │  Server opens envelope with        │
  │  X25519(server_priv, eph_pub)      │
  │  Verifies MAC                      │
  │  Checks nonce + timestamp          │
  │  Processes request                 │
  │  Seals response + Ed25519 sign     │
  │                                    │
  │  Response:                         │
  │    {"e": ct, "m": mac, "ek": eph, │
  │     "s": ed25519_signature}        │
  │  <───────────────────────────────── │
  │                                    │
  │  Verify Ed25519 signature          │
  │  Open response envelope            │
  │  Check timestamp (30s window)      │

Key Derivation

All keys are derived via HKDF-SHA256:

Purpose Salt Info Source
Master key (AES encrypt/decrypt) keyvory-kdf-v1 master-key SECRET_KEY string
AES encryption (envelope) keyvory-x25519-v1 aes-from-x25519 X25519 shared secret
MAC signing (envelope) keyvory-x25519-v1 mac-from-x25519 X25519 shared secret

AES-GCM Format

Encrypted data is: base64(nonce[12] + ciphertext + tag[16])

Anti-Tamper

The SDK includes several layers of protection:

Module integrity — on import, the SDK hashes its own source file with SHA-256 and stores the hash in _SELF_INTEGRITY. This provides a reference hash that can be checked by external integrity verification tools.

Debugger detection — sys.gettrace() is checked before every sensitive operation (initialize, login, heartbeat, get_license_info). A non-None trace function (attached by pdb, pydevd, or any profiler) raises TamperError.

Timing detection — a fixed SHA-256 computation is timed using time.perf_counter_ns(). If it takes longer than 5ms (a generous threshold that covers slow VMs and Windows), a TamperError is raised. This catches single-step debuggers and emulators.

HMAC state chain — the session key, key version, initialization flag, and X25519 private key are chained into an HMAC-SHA256 computed from the shared secret:

state_mac = HMAC-SHA256(
    secret_key,
    "{client_key}:{key_version}:{is_initialized}:{x25519_private_key}"
)

Verified before every sensitive call via _verify_state_mac(). Updated after state changes via _update_state_mac(). Any in-memory modification to these fields is detected and raises TamperError.

These checks protect against casual tampering. They are not a substitute for server-side validation — always treat the server as the ground truth.

Key Rotation

The AES session key rotates automatically every 100 heartbeat calls via _maybe_rotate(). You do not need to manage this.

If you call heartbeat() infrequently (e.g., once per hour), rotation may never trigger in a short-lived process. That is fine — the session key is strong (256-bit random). Rotation is an extra hardening measure for long-running services.

You can also force a rotation at any time:

if not client.rotate_key():
    # rotation failed — the old key is still in use
    logger.warning("Key rotation failed")

Error Handling

A production integration should handle all three failure modes:

import os, sys, time
from keyvory_client import KeyvoryClient, TamperError, NetworkError

def start_app(license_key: str):
    with KeyvoryClient(
        server_url=os.environ["KEYVORY_SERVER"],
        secret_key=os.environ["KEYVORY_SECRET"],
        max_retries=3,
        timeout=10,
    ) as client:

        try:
            initialized = client.initialize()
        except TamperError:
            print("ERROR: Integrity check failed.")
            sys.exit(1)
        except NetworkError as e:
            print(f"ERROR: Cannot reach license server: {e}")
            sys.exit(1)

        if not initialized:
            print("ERROR: License server initialization failed. Check URL and secret key.")
            sys.exit(1)

        result = client.login(license_key)
        if not result.is_valid:
            print(f"ERROR: License invalid — {result.status} ({result.code})")
            sys.exit(1)

        print(f"License valid. Expires {result.expiration_date}.")
        run_main_loop(client, license_key)


def run_main_loop(client, license_key):
    while True:
        try:
            alive = client.heartbeat(license_key=license_key)
            if not alive:
                print("License revoked. Shutting down.")
                sys.exit(1)
        except NetworkError:
            # Don't kill the app on a transient network hiccup.
            # The SDK already retried 3 times.
            print("WARNING: Heartbeat failed. Will retry next cycle.")
        time.sleep(60)

Examples

Gate a feature flag

result = client.login(license_key)
if result.is_valid and result.has_feature("api_access"):
    return serve_api_request(request)
else:
    return error_403("API access not included in your license.")

Display license info in a settings screen

info = client.get_license_info(license_key)

settings_panel.set({
    "license_type":  info.license_type,
    "expires":       info.expiration_date,
    "max_devices":   info.max_devices,
    "features":      [k for k, v in info.features.items() if v],
})

Background heartbeat thread

import threading, time, sys, os

stop_event = threading.Event()

def heartbeat_loop(client, key):
    while not stop_event.wait(timeout=60):
        try:
            if not client.heartbeat(license_key=key):
                print("License revoked. Exiting.")
                os._exit(1)
        except NetworkError:
            pass   # transient; try again next cycle

thread = threading.Thread(target=heartbeat_loop, args=(client, license_key), daemon=True)
thread.start()

# ... run app ...

stop_event.set()

Force a key rotation

# Manual rotation (rarely needed — happens every 100 heartbeats automatically)
if client.rotate_key():
    print("Key rotated successfully")
else:
    print("Rotation failed, continuing with current key")

Running the bundled demo

The example/python/example_client.py script walks through all five SDK operations interactively:

cd example/python
KEYVORY_SERVER=http://localhost:5000 \
KEYVORY_SECRET=your-secret-key \
python example_client.py

You will be prompted to enter a license key. The demo walks through five steps: init handshake, license validation, info query, heartbeat keepalive, and session key rotation.