Browse the library

Documentation libraryGet started by language

Write your own client

When your language has no SDK, cannot call C functions and cannot run the command line tool, you can write your own client for the CLV1.1 protocol. This guide goes through what a client must do, with a sample client written in plain Python from the specification alone, run against the real API in CI, and the test vectors to check your own client with.

5 min read

Do you really need to?

Weigh the other ways first: the C library can be called from most languages, and the command line tool works anywhere a program can be started. Writing your own client means getting signing and signature checks exactly right yourself; one wrong detail and your software refuses every answer, or worse, trusts a forged one.

What a client must do

Keep an Ed25519 device key on the machine, made once and reused on every run.
Sign every request with the device key, with a single-use nonce and a timestamp.
Check every answer, error answers included, with the app's public key: the signature must be bound to the request just sent and be no more than 60 seconds off.
Send heartbeats while the software runs, at the pace the server signs as nextCheckSeconds, and sign out when it closes.
Try again only when it is safe, as the questions below explain.

The whole specification (headers, signed strings, endpoints, offline tokens, error codes) is in the CLV1 protocol specification. Below, each step is shown with code that runs.

1. Keys

The app's public key is on the Integration › Connect software page. Its kid is the first 16 hex characters of SHA-256 over the key's 32 bytes; an answer signed by a key with another kid is refused. The device key is made once and kept: sessions and offline tokens are bound to it, while the seat on the license follows the machine's hashed identifiers; the machine's identifier is hashed on the machine:

# The app's public key, from the dashboard. Its kid is the first 16 hex characters of sha256 of its 32 bytes.
APP_KEY_BYTES = from_b64url(os.environ["CLAVIS_APP_KEY"])
APP_KEY = Ed25519PublicKey.from_public_bytes(APP_KEY_BYTES)
APP_KID = sha256_hex(APP_KEY_BYTES)[:16]


def device_key() -> Ed25519PrivateKey:
    """Made once and kept: the same key every run, so sessions and offline tokens stay with this machine."""
    path = STATE / "raw-device.key"
    if path.exists():
        return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(path.read_text()))
    key = Ed25519PrivateKey.generate()
    STATE.mkdir(parents=True, exist_ok=True)
    path.write_text(key.private_bytes_raw().hex())
    return key


DEVICE_KEY = device_key()


def fingerprint() -> str:
    """Hashed on the machine: sha256(lowercase(trim(value))), first 32 hex characters. Raw values never leave."""
    machine_id = Path("/etc/machine-id")
    value = machine_id.read_text() if machine_id.exists() else platform.node()
    return sha256_hex(value.strip().lower().encode())[:32]

2. Sign the request

Each request is a POST to /v1/client/<endpoint> with a JSON body. The signed string has eight lines: CLV1-REQUEST, the App ID, the method, the path, SHA-256 of the exact body bytes, the timestamp, the nonce and the session id (or -). The x-clavis-proto: 1.1 header asks for CLV1.1 answers. The x-clavis-sdk header is not signed; it only lets the Connect software page show your client's name:

def call(endpoint: str, body: dict, session: str = None, retry: bool = True):
    """POST one signed request; returns (HTTP status, answer) once the answer's signature is checked."""
    global clock_offset
    raw = json.dumps(body, separators=(",", ":")).encode()
    path = "/v1/client/" + endpoint
    timestamp = str(int(time.time()) + clock_offset)
    nonce = secrets.token_urlsafe(18)
    signed = "\n".join(["CLV1-REQUEST", APP_ID, "POST", path, sha256_hex(raw), timestamp, nonce, session or "-"])
    headers = {
        "content-type": "application/json",
        "x-clv-app": APP_ID,
        "x-clv-time": timestamp,
        "x-clv-nonce": nonce,
        "x-clv-key": b64url(DEVICE_KEY.public_key().public_bytes_raw()),
        "x-clv-signature": b64url(DEVICE_KEY.sign(signed.encode())),
        "x-clavis-proto": "1.1",
        "x-clavis-sdk": "raw-python/1.0.0 (%s)" % platform.system().lower(),
    }
    if session:
        headers["x-clv-session"] = session
    request = urllib.request.Request(BASE_URL + path, data=raw, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            status, answer, answer_headers = response.status, response.read(), response.headers
    except urllib.error.HTTPError as err:
        status, answer, answer_headers = err.code, err.read(), err.headers
    except OSError:
        fail("network_error")
    verify(status, path, nonce, answer, answer_headers)
    result = json.loads(answer)
    # A signed clock_skew says this machine's clock is off: correct it and try once more.
    if result.get("error", {}).get("code") == "clock_skew" and retry:
        clock_offset = result["serverTime"] - int(time.time())
        return call(endpoint, body, session, retry=False)
    return status, result

3. Check the answer

A CLV1.1 answer carries three headers, x-clavis-kid, x-clavis-timestamp and x-clavis-signature. The signed string has eight lines: CLV1.1, the kid, the HTTP status, the method, the path, the request's nonce, the timestamp and SHA-256 of the body. A missing header, another kid, a wrong signature or more than 60 seconds off: treat it as no answer at all.

def verify(status: int, path: str, nonce: str, body: bytes, headers) -> None:
    """The verification contract: every answer, error answers included, or it is no answer at all."""
    kid = headers.get("x-clavis-kid")
    timestamp = headers.get("x-clavis-timestamp")
    signature = headers.get("x-clavis-signature")
    if not (kid and timestamp and signature):
        fail("unsigned_response")
    if kid != APP_KID:
        fail("unknown_key")
    signed = "\n".join(["CLV1.1", kid, str(status), "POST", path, nonce, timestamp, sha256_hex(body)])
    try:
        APP_KEY.verify(from_b64url(signature), signed.encode())
    except InvalidSignature:
        fail("invalid_response_signature")
    if abs(int(timestamp) - (time.time() + clock_offset)) > 60:
        fail("stale_response")

Each case has its own code (unsigned_response, unknown_key, invalid_response_signature, stale_response), the same as the official SDKs, see Error codes.

4. K1 from start to end

init, then activate with the key and the machine's description, heartbeat with the session id just received, and finally logout:

call("init", {})
status, grant = call("activate", {"key": os.environ["CLAVIS_LICENSE"], "device": {"fingerprint": fingerprint()}})
if status != 200:
    fail(grant["error"]["code"])
session = grant["session"]["id"]
# Every few minutes while the software runs; sessions end after an hour without one.
status, beat = call("heartbeat", {}, session)
if status != 200:
    fail(beat["error"]["code"])
call("logout", {}, session)
print("OK k1", grant["license"]["status"])

The whole file needs only Python 3.9 or later and the cryptography package:

"""K1 without an SDK: a client written from /docs/protocol alone, in plain Python with the
`cryptography` package for Ed25519. It shows what a client in any language must do: sign every request, check every
answer, keep the device key.

Settings: CLAVIS_BASE_URL, CLAVIS_APP_ID, CLAVIS_APP_KEY, CLAVIS_LICENSE, CLAVIS_STATE_DIR.
Ends with "OK k1 active", or "FAIL <code>".
"""

import base64
import hashlib
import json
import os
import platform
import secrets
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey

BASE_URL = os.environ["CLAVIS_BASE_URL"].rstrip("/")
APP_ID = os.environ["CLAVIS_APP_ID"]
STATE = Path(os.environ.get("CLAVIS_STATE_DIR", ".clavis"))


def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def from_b64url(text: str) -> bytes:
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def fail(code: str) -> None:
    print("FAIL", code)
    sys.exit(1)


# The app's public key, from the dashboard. Its kid is the first 16 hex characters of sha256 of its 32 bytes.
APP_KEY_BYTES = from_b64url(os.environ["CLAVIS_APP_KEY"])
APP_KEY = Ed25519PublicKey.from_public_bytes(APP_KEY_BYTES)
APP_KID = sha256_hex(APP_KEY_BYTES)[:16]


def device_key() -> Ed25519PrivateKey:
    """Made once and kept: the same key every run, so sessions and offline tokens stay with this machine."""
    path = STATE / "raw-device.key"
    if path.exists():
        return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(path.read_text()))
    key = Ed25519PrivateKey.generate()
    STATE.mkdir(parents=True, exist_ok=True)
    path.write_text(key.private_bytes_raw().hex())
    return key


DEVICE_KEY = device_key()


def fingerprint() -> str:
    """Hashed on the machine: sha256(lowercase(trim(value))), first 32 hex characters. Raw values never leave."""
    machine_id = Path("/etc/machine-id")
    value = machine_id.read_text() if machine_id.exists() else platform.node()
    return sha256_hex(value.strip().lower().encode())[:32]



clock_offset = 0


def call(endpoint: str, body: dict, session: str = None, retry: bool = True):
    """POST one signed request; returns (HTTP status, answer) once the answer's signature is checked."""
    global clock_offset
    raw = json.dumps(body, separators=(",", ":")).encode()
    path = "/v1/client/" + endpoint
    timestamp = str(int(time.time()) + clock_offset)
    nonce = secrets.token_urlsafe(18)
    signed = "\n".join(["CLV1-REQUEST", APP_ID, "POST", path, sha256_hex(raw), timestamp, nonce, session or "-"])
    headers = {
        "content-type": "application/json",
        "x-clv-app": APP_ID,
        "x-clv-time": timestamp,
        "x-clv-nonce": nonce,
        "x-clv-key": b64url(DEVICE_KEY.public_key().public_bytes_raw()),
        "x-clv-signature": b64url(DEVICE_KEY.sign(signed.encode())),
        "x-clavis-proto": "1.1",
        "x-clavis-sdk": "raw-python/1.0.0 (%s)" % platform.system().lower(),
    }
    if session:
        headers["x-clv-session"] = session
    request = urllib.request.Request(BASE_URL + path, data=raw, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            status, answer, answer_headers = response.status, response.read(), response.headers
    except urllib.error.HTTPError as err:
        status, answer, answer_headers = err.code, err.read(), err.headers
    except OSError:
        fail("network_error")
    verify(status, path, nonce, answer, answer_headers)
    result = json.loads(answer)
    # A signed clock_skew says this machine's clock is off: correct it and try once more.
    if result.get("error", {}).get("code") == "clock_skew" and retry:
        clock_offset = result["serverTime"] - int(time.time())
        return call(endpoint, body, session, retry=False)
    return status, result




def verify(status: int, path: str, nonce: str, body: bytes, headers) -> None:
    """The verification contract: every answer, error answers included, or it is no answer at all."""
    kid = headers.get("x-clavis-kid")
    timestamp = headers.get("x-clavis-timestamp")
    signature = headers.get("x-clavis-signature")
    if not (kid and timestamp and signature):
        fail("unsigned_response")
    if kid != APP_KID:
        fail("unknown_key")
    signed = "\n".join(["CLV1.1", kid, str(status), "POST", path, nonce, timestamp, sha256_hex(body)])
    try:
        APP_KEY.verify(from_b64url(signature), signed.encode())
    except InvalidSignature:
        fail("invalid_response_signature")
    if abs(int(timestamp) - (time.time() + clock_offset)) > 60:
        fail("stale_response")



call("init", {})
status, grant = call("activate", {"key": os.environ["CLAVIS_LICENSE"], "device": {"fingerprint": fingerprint()}})
if status != 200:
    fail(grant["error"]["code"])
session = grant["session"]["id"]
# Every few minutes while the software runs; sessions end after an hour without one.
status, beat = call("heartbeat", {}, session)
if status != 200:
    fail(beat["error"]["code"])
call("logout", {}, session)
print("OK k1", grant["license"]["status"])

Run it with a test key (the Make a test key button):

pip install cryptography
export CLAVIS_BASE_URL="<API address>"
export CLAVIS_APP_ID="<App ID>"
export CLAVIS_APP_KEY="<public key>"
export CLAVIS_LICENSE="<test key>"
python k1_raw.py

The last line reads OK k1 active, and the Connect software page shows Connected with raw-python/1.0.0. CI runs this file against the real API on Windows and Linux: it activates, a key that cannot be used gets FAIL invalid_credentials, and when another app's public key is configured by mistake, the client refuses the answer with FAIL unknown_key.

5. Check yourself with the test vectors

Download clv1.1-test-vectors.json from Downloads. It holds keys, signed strings, signatures, an offline token and a release manifest, with the results the server and every official SDK reach. Write the checks below with your own client's code; once they all pass, your client signs and verifies exactly as the server does. The Python version of the checks:

base64url without padding, lowercase hex SHA-256, and how hardware identifiers are hashed:

# base64url without padding, SHA-256 in lowercase hex, and hardware identifiers hashed before they leave.
for v in vectors["base64url"]:
    check("base64url " + v["hex"], b64url(bytes.fromhex(v["hex"])) == v["b64"])
for v in vectors["sha256"]:
    check("sha256 " + v["hex"], sha256_hex(v["text"].encode()) == v["hex"])
for v in vectors["hashComponent"]:
    check("hashComponent", sha256_hex(v["input"].strip().lower().encode())[:32] == v["output"])

Keys from their seeds, the kid and the device key hash:

keys = vectors["keys"]
app_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["appSeedHex"])).public_key()
device_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["deviceSeedHex"])).public_key()
check("app public key from its seed", b64url(app_key.public_bytes_raw()) == keys["appPublicKey"])
check("kid", sha256_hex(app_key.public_bytes_raw())[:16] == keys["kid"])
check("device public key from its seed", b64url(device_key.public_bytes_raw()) == keys["devicePublicKey"])
check("device key hash", sha256_hex(device_key.public_bytes_raw()) == keys["deviceKeyHash"])

A request's signed string and signature:

# The string the device signs, and the signature the server expects for it.
r = vectors["request"]
lines = ["CLV1-REQUEST", r["appId"], r["method"], r["path"], sha256_hex(r["body"].encode())]
signed = "\n".join(lines + [str(r["timestamp"]), r["nonce"], r["sessionId"] or "-"])
check("request string", signed == r["canonical"])
device = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["deviceSeedHex"]))
check("request signature", b64url(device.sign(signed.encode())) == r["signature"])

An answer's signed string and signature:

# The string the server signs for a 1.1 client: verify it with the app's public key.
s = vectors["response"]
lines = ["CLV1.1", s["kid"], str(s["status"]), s["method"], s["path"], s["requestNonce"], str(s["timestamp"])]
signed = "\n".join(lines + [sha256_hex(s["body"].encode())])
check("response string", signed == s["canonical"])
app_public = Ed25519PublicKey.from_public_bytes(from_b64url(keys["appPublicKey"]))
check("response signature", signed_by(app_public, s["signature"], signed))

The offline token: clv1.<payload>.<signature>, where the app signs clv1.<payload>; it is valid only on the machine whose key hash matches dev, from nbf to exp, and refused when the machine's clock was turned back:

# clv1.<payload>.<signature>: the app key signs "clv1.<payload>"; the payload names the device by its key hash.
o = vectors["offline"]
prefix, payload, signature = o["token"].split(".")
check("offline signature", signed_by(app_key, signature, prefix + "." + payload))
claims = json.loads(from_b64url(payload))
check("offline payload", claims == o["payload"] and claims["dev"] == keys["deviceKeyHash"])
check("offline valid", claims["nbf"] <= o["validAt"] < claims["exp"])
check("offline expired", o["expiredAt"] >= claims["exp"])
check("offline clock turned back", o["clockBackAt"] < claims["iat"])

The release manifest: the app signs CLV1-MANIFEST, a line break, then the manifest text exactly as received:

# A release manifest: the app key signs "CLV1-MANIFEST\n" followed by the manifest text exactly as received.
m = vectors["manifest"]
check("manifest signature", signed_by(app_key, m["signature"], "CLV1-MANIFEST\n" + m["manifest"]))
manifest = json.loads(m["manifest"])
check("manifest kid", manifest["kid"] == m["kid"])
check("manifest valid", manifest["issuedAt"] <= m["validAt"] < manifest["expiresAt"] <= m["expiredAt"])

The whole Python file, with the vectors it loads and its check and signed_by helpers:

"""The compatibility test for your own client: every check of clv1.1-test-vectors.json (the
Downloads page), done here in plain Python with `cryptography` for Ed25519. Write the same checks with your own
client's code; when they all pass, it signs and verifies exactly as the server and every official SDK do.

Run: python check_vectors.py clv1.1-test-vectors.json      Ends with "OK vectors", or the first check that failed.
"""

import base64
import hashlib
import json
import sys

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey


def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def from_b64url(text: str) -> bytes:
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def signed_by(key: Ed25519PublicKey, signature: str, data: str) -> bool:
    try:
        key.verify(from_b64url(signature), data.encode())
        return True
    except InvalidSignature:
        return False


def check(name: str, ok: bool) -> None:
    if not ok:
        print("FAIL", name)
        sys.exit(1)


vectors = json.load(open(sys.argv[1] if len(sys.argv) > 1 else "clv1.1-test-vectors.json", encoding="utf-8"))

# base64url without padding, SHA-256 in lowercase hex, and hardware identifiers hashed before they leave.
for v in vectors["base64url"]:
    check("base64url " + v["hex"], b64url(bytes.fromhex(v["hex"])) == v["b64"])
for v in vectors["sha256"]:
    check("sha256 " + v["hex"], sha256_hex(v["text"].encode()) == v["hex"])
for v in vectors["hashComponent"]:
    check("hashComponent", sha256_hex(v["input"].strip().lower().encode())[:32] == v["output"])

keys = vectors["keys"]
app_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["appSeedHex"])).public_key()
device_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["deviceSeedHex"])).public_key()
check("app public key from its seed", b64url(app_key.public_bytes_raw()) == keys["appPublicKey"])
check("kid", sha256_hex(app_key.public_bytes_raw())[:16] == keys["kid"])
check("device public key from its seed", b64url(device_key.public_bytes_raw()) == keys["devicePublicKey"])
check("device key hash", sha256_hex(device_key.public_bytes_raw()) == keys["deviceKeyHash"])

# The string the device signs, and the signature the server expects for it.
r = vectors["request"]
lines = ["CLV1-REQUEST", r["appId"], r["method"], r["path"], sha256_hex(r["body"].encode())]
signed = "\n".join(lines + [str(r["timestamp"]), r["nonce"], r["sessionId"] or "-"])
check("request string", signed == r["canonical"])
device = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(keys["deviceSeedHex"]))
check("request signature", b64url(device.sign(signed.encode())) == r["signature"])

# The string the server signs for a 1.1 client: verify it with the app's public key.
s = vectors["response"]
lines = ["CLV1.1", s["kid"], str(s["status"]), s["method"], s["path"], s["requestNonce"], str(s["timestamp"])]
signed = "\n".join(lines + [sha256_hex(s["body"].encode())])
check("response string", signed == s["canonical"])
app_public = Ed25519PublicKey.from_public_bytes(from_b64url(keys["appPublicKey"]))
check("response signature", signed_by(app_public, s["signature"], signed))

# clv1.<payload>.<signature>: the app key signs "clv1.<payload>"; the payload names the device by its key hash.
o = vectors["offline"]
prefix, payload, signature = o["token"].split(".")
check("offline signature", signed_by(app_key, signature, prefix + "." + payload))
claims = json.loads(from_b64url(payload))
check("offline payload", claims == o["payload"] and claims["dev"] == keys["deviceKeyHash"])
check("offline valid", claims["nbf"] <= o["validAt"] < claims["exp"])
check("offline expired", o["expiredAt"] >= claims["exp"])
check("offline clock turned back", o["clockBackAt"] < claims["iat"])

# A release manifest: the app key signs "CLV1-MANIFEST\n" followed by the manifest text exactly as received.
m = vectors["manifest"]
check("manifest signature", signed_by(app_key, m["signature"], "CLV1-MANIFEST\n" + m["manifest"]))
manifest = json.loads(m["manifest"])
check("manifest kid", manifest["kid"] == m["kid"])
check("manifest valid", manifest["issuedAt"] <= m["validAt"] < manifest["expiresAt"] <= m["expiredAt"])

print("OK vectors")

Run it with python check_vectors.py clv1.1-test-vectors.json (it needs pip install cryptography); the last line is OK vectors. Change one character of any signature in the file and it stops at that check.

Questions

**Must I send x-clavis-proto: 1.1?** Yes. Without it the server answers as in the older CLV1, signing fewer fields; a new client should always use 1.1.

What if the customer's clock is wrong? The server answers clock_skew (signed) with serverTime; the client works out the offset, adds it to its timestamps and sends once more, as the sample in step 2 does.

When should my client try again? With no answer or an unsigned 5xx page, repeat only init, activate, login, heartbeat, variables, updates, logout and report. Never repeat trial, register, redeem or transfer after a lost answer: they may already have happened. A signed rate_limited, or an unsigned 429 or 503, can be repeated on any call, after the wait in the Retry-After header. Wait a growing, random time between tries, and sign each try again with a new nonce. Only a verified answer changes license state.

How often do I send heartbeats? Sign-in and heartbeat answers carry a signed nextCheckSeconds. Wait that long, plus a spread of up to a fifth that stays the same for the machine, so thousands of copies do not call at once.

And trials, accounts, variables, updates? The same signing and checking, with other endpoints and bodies: the endpoint table is in the CLV1 protocol specification, and each scenario's calls are in How the system works.