ETSI 014 client (qkdsec.client)

ETSI GS QKD 014 REST client for Key Management Entity (KME) systems.

Use this to fetch quantum keys from real QKD hardware (Toshiba, ID Quantique, QuantumCTek, etc.) or from a local KME simulator. Both synchronous and asynchronous clients are provided.

Sync example:
>>> from qkdsec.client import ETSI014Client
>>> kme = ETSI014Client("https://kme.example.com",
...                     client_cert=("alice.crt", "alice.key"))
>>> status = kme.status("sae-bob")
>>> keys = kme.get_enc_keys("sae-bob", number=1, size=256)

Async example (requires pip install qkdsec[async]):

from qkdsec.client.aio import AsyncETSI014Client

async with AsyncETSI014Client("https://kme.example.com",
                              client_cert=("alice.crt", "alice.key")) as kme:
    keys = await kme.get_enc_keys("sae-bob", number=1, size=256)

Synchronous client

class qkdsec.client.ETSI014Client(base_url, *, client_cert=None, verify=True, timeout=30.0, extra_headers=None, session=None)[source]

Bases: object

Synchronous client for an ETSI GS QKD 014 Key Management Entity (KME).

Parameters:

  • base_url (str) – Base URL of the KME (e.g., "https://kme.example.com:8443"). The /api/v1/keys path is appended automatically.

  • client_cert (Union[str, tuple[str, str], None]) – Client certificate for mTLS. Pass a path to a combined cert+key PEM file, or a (cert_path, key_path) tuple. Most production KMEs require mTLS.

  • verify (Union[bool, str]) – TLS verification mode. True (default) uses the system CA bundle. Pass a path to a CA certificate file to pin a custom CA. False disables verification (do not use in production).

  • timeout (float) – Per-request timeout in seconds. Default 30.

  • extra_headers (Optional[dict]) – Additional headers to send on every request.

  • session (Optional[Session]) – Reuse an existing session (useful for connection pooling or testing).

close()[source]

Close the underlying HTTP session.

Return type:

None

get_dec_keys(slave_sae_id, *, key_ids, key_id_extensions=None, key_ids_extension=None)[source]

Fetch specific keys by key_ID for the slave SAE (ETSI 014 §5.4).

Returns just the list of keys. Use get_dec_keys_container() to access key_container_extension.

Return type:

list[KeyResponse]

Parameters:
get_dec_keys_container(slave_sae_id, *, key_ids, key_id_extensions=None, key_ids_extension=None)[source]

Fetch keys by key_ID and return the full ETSI 014 §5.4 container.

Parameters:

  • slave_sae_id (str) – The SAE ID of the decrypting peer (this client).

  • key_ids (list[str]) – The key identifiers to retrieve, previously obtained by the master SAE via get_enc_keys.

  • key_id_extensions (Optional[dict[str, dict[str, Any]]]) – Per-key extension data, keyed by key_ID (ETSI 014 §5.4.2 key_ID_extension).

  • key_ids_extension (Optional[dict[str, Any]]) – Container-level extension data for the request (ETSI 014 §5.4.2 key_IDs_extension).

Return type:

KeysContainer

get_enc_keys(slave_sae_id, *, number=1, size=256, method='GET', additional_slave_sae_ids=None, extension_mandatory=None, extension_optional=None)[source]

Fetch encryption keys for the master SAE (ETSI 014 §5.3).

Returns just the list of keys for backwards compatibility. Use get_enc_keys_container() to access key_container_extension.

Return type:

list[KeyResponse]

Parameters:
get_enc_keys_container(slave_sae_id, *, number=1, size=256, method='GET', additional_slave_sae_ids=None, extension_mandatory=None, extension_optional=None)[source]

Fetch encryption keys and return the full ETSI 014 §5.3 container.

Parameters:

  • slave_sae_id (str) – The SAE ID of the intended decrypting peer.

  • number (int) – Number of keys to request. KMEs typically cap this at 20.

  • size (int) – Key size in bits. Must be a multiple of 8.

  • method (str) – "GET" (query params) or "POST" (JSON body). Multicast and extensions force POST automatically since GET cannot carry them.

  • additional_slave_sae_ids (Optional[list[str]]) – Additional slave SAEs that should be able to retrieve the same key (ETSI 014 multicast). Forces POST.

  • extension_mandatory (Optional[list[dict[str, Any]]]) – Vendor-specific parameters the KME MUST honor (ETSI 014 §5.3.2). Forces POST. The KME rejects the request if it cannot satisfy.

  • extension_optional (Optional[list[dict[str, Any]]]) – Vendor-specific parameters the KME MAY honor (ETSI 014 §5.3.2). Forces POST. Best-effort.

Return type:

KeysContainer

status(slave_sae_id)[source]

Fetch KME status for the given slave SAE (ETSI 014 §5.2).

Return type:

StatusResponse

Parameters:

slave_sae_id (str)

Asynchronous client

class qkdsec.client.aio.AsyncETSI014Client(base_url, *, client_cert=None, verify=True, timeout=30.0, extra_headers=None, client=None)[source]

Bases: object

Asynchronous client for an ETSI GS QKD 014 KME.

Parameters mirror qkdsec.client.ETSI014Client exactly.

Use as an async context manager to ensure the underlying httpx client is closed:

async with AsyncETSI014Client(...) as kme:
    await kme.status("sae-bob")
Parameters:
async aclose()[source]

Close the underlying HTTP client (if owned by this instance).

Return type:

None

async get_dec_keys(slave_sae_id, *, key_ids, key_id_extensions=None, key_ids_extension=None)[source]

Fetch specific keys by key_ID for the slave SAE (ETSI 014 §5.4).

Return type:

list[KeyResponse]

Parameters:
async get_dec_keys_container(slave_sae_id, *, key_ids, key_id_extensions=None, key_ids_extension=None)[source]

Fetch keys by key_ID and return the full ETSI 014 §5.4 container.

Return type:

KeysContainer

Parameters:
async get_enc_keys(slave_sae_id, *, number=1, size=256, method='GET', additional_slave_sae_ids=None, extension_mandatory=None, extension_optional=None)[source]

Fetch encryption keys for the master SAE (ETSI 014 §5.3).

Return type:

list[KeyResponse]

Parameters:
async get_enc_keys_container(slave_sae_id, *, number=1, size=256, method='GET', additional_slave_sae_ids=None, extension_mandatory=None, extension_optional=None)[source]

Fetch encryption keys and return the full ETSI 014 §5.3 container.

Return type:

KeysContainer

Parameters:
async status(slave_sae_id)[source]

Fetch KME status for the given slave SAE (ETSI 014 §5.2).

Return type:

StatusResponse

Parameters:

slave_sae_id (str)

Response types

class qkdsec.client.KeyResponse(key_id, key, key_id_extension=None, key_extension=None)[source]

Bases: object

A single key fetched from the KME (ETSI 014 §5.3.3 / §5.4.3).

Parameters:
key_id

The KME-assigned key identifier (UUID). Used to retrieve the matching key on the slave SAE via get_dec_keys.

Type:

str

key

The raw key material. The wire format is base64; this field is already decoded.

Type:

bytes

key_id_extension

Vendor-specific or non-standard extension data attached to the key identifier (ETSI 014 §5.3.3 key_ID_extension).

Type:

dict, optional

key_extension

Vendor-specific or non-standard extension data attached to the key itself (ETSI 014 §5.3.3 key_extension).

Type:

dict, optional

key: bytes
key_extension: dict[str, Any] | None = None
key_id: str
key_id_extension: dict[str, Any] | None = None
property size_bits: int
class qkdsec.client.KeysContainer(keys=<factory>, key_container_extension=None)[source]

Bases: object

A container of keys returned by enc_keys / dec_keys (ETSI 014 §5.3.3).

Wraps list[KeyResponse] to also surface the optional container-level extension. Iterating over a KeysContainer yields its keys, so most callers can treat it like a list.

Parameters:
keys
Type:

list[KeyResponse]

key_container_extension

Vendor-specific extension data attached to the container as a whole (ETSI 014 §5.3.3 key_container_extension).

Type:

dict, optional

key_container_extension: dict[str, Any] | None = None
keys: list[KeyResponse]
class qkdsec.client.StatusResponse(source_kme_id, target_kme_id, master_sae_id, slave_sae_id, key_size, stored_key_count, max_key_count, max_key_per_request, max_key_size, min_key_size, max_sae_id_count, status_extension=None)[source]

Bases: object

KME status for a given slave SAE (ETSI GS QKD 014 §5.2).

Parameters:

  • source_kme_id (str)

  • target_kme_id (str)

  • master_sae_id (str)

  • slave_sae_id (str)

  • key_size (int)

  • stored_key_count (int)

  • max_key_count (int)

  • max_key_per_request (int)

  • max_key_size (int)

  • min_key_size (int)

  • max_sae_id_count (int)

  • status_extension (dict[str, Any] | None)

key_size: int
master_sae_id: str
max_key_count: int
max_key_per_request: int
max_key_size: int
max_sae_id_count: int
min_key_size: int
slave_sae_id: str
source_kme_id: str
status_extension: dict[str, Any] | None = None
stored_key_count: int
target_kme_id: str

Errors

exception qkdsec.client.KMEError[source]

Bases: Exception

Base class for all KME client errors.

exception qkdsec.client.KMEHTTPError(status_code, message)[source]

Bases: KMEError

Raised when the KME returns a non-2xx HTTP response.

Parameters:

  • status_code (int)

  • message (str)

status_code
Type:

int

message

The KME’s error message (or response body), if available.

Type:

str

exception qkdsec.client.KMENotFoundError(message)[source]

Bases: KMEHTTPError

Raised on HTTP 404 (e.g., key_ID not found or already retrieved).

Parameters:

message (str)