"""Make a DKE store key on your own machine, from Python. import os from pathlib import Path from dke.keys import mint made = mint("notes", label="nightly", level="read", token=os.environ["DKEKEY_TOKEN"], open_with=Path("notes.key").read_text()) fd = os.open("nightly.key", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as out: out.write(made.key + "\n") This is what the `dkekey` tool does, as a library. The key is made here, from the operating system's random source. To add it to a store, this module opens the store with a key that already opens it, wraps the store's data key under the new key, and hands the account service two things: the new key's SHA-256 hash and that wrap. The key itself is never sent, and neither is the key that opened the store, nor a password: a password derives, here, the proof the account checks and the key the account's customer key is kept under, exactly as the dashboard does. AN EXTRA, NOT THE CORE. The `keys` extra adds one dependency, `cryptography`, and this module uses it for one thing: AES-GCM, which the standard library does not have. HKDF, PBKDF2, SHA-256 and the random source are the standard library's. The rest of the package stays dependency-free: `import dke` never reaches this module, which the test suite asserts by reading the import graph and again by running it. A NETWORK MODULE, LIKE `dke.client`. It talks to the account service, it is reached by nothing the producing half imports, and importing it is the grant. SIGNING IN, one of two ways: * `email=` and `password=`, as a person signs in to the dashboard. With nothing else given, the store is opened with the account's customer key, which the password opens. * `token=`, an account token made in the dashboard. A token can only add keys to stores that exist, up to the level it was made with, and opens no store by itself: give `open_with=` as well. `open_with=` is a key that already opens the store: one of its store keys (`dke_s_...`) or the account's customer key (`dke_c_...`). FAILURES. A wrong argument raises `ValueError` before anything is sent. Anything the account service refuses, or a key that does not open the store, raises `MintError`, a `dke.client.DkeError`, whose message says what to do. """ from __future__ import annotations import base64 import hashlib import hmac import json import re import secrets import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from typing import Any try: from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import AESGCM except ImportError as missing: # the extra is not installed raise ImportError( "dke.keys needs the cryptography package, for AES-GCM, which the 'keys' " "extra adds: pip install 'cryptography>=38'") from missing from dke import __version__ from dke.client import DkeError, _NoRedirect __all__ = ["mint", "MintedKey", "MintError", "LEVELS", "KEY_FORMAT", "DEFAULT_API"] DEFAULT_API = "https://dke.langsyn.com/api/account" # The format a key's wrap is made in, sent with every key. The account service # cannot open a wrap to check it, so a copy of this module left behind by a # change of format is refused by name rather than leaving a key that opens # nothing. KEY_FORMAT = "dke-store-kek-v2" LEVELS = ("read", "read-add", "read-write", "read-write-delete") _CUSTOMER_SEAL_INFO = "dke-store-customer-kek-v1" _LOGIN_PROOF_INFO = "langsyn-login-proof-v1" _CUSTOMER_KEY_INFO = "langsyn-customer-key-v1" _PBKDF2_ITERATIONS = 600000 _KEY_LEN = 32 _NONCE_LEN = 12 _TAG_LEN = 16 _STORE_KEY_PREFIX = "dke_s_" _TOKEN_PREFIX = "dke_a_" _SESSION_COOKIE = "ls_session" _CUSTOMER_KEY_TEXT = re.compile(r"dke_[ck]_([A-Za-z0-9_-]{43})") # JavaScript's white space and line terminators, which is what the dashboard # trims an address by. Not str.strip()'s set: that trims U+0085 and the # information separators, and leaves U+FEFF. _JS_SPACE = ("\t\n\v\f\r         " "     

  　") class MintError(DkeError): """The key was not made: refused by the account service, or the store would not open with what was given. `status` is the account service's HTTP status and `code` its refusal code, when it answered with them.""" def __init__(self, message: str, *, status: int | None = None, code: str | None = None) -> None: super().__init__(message) self.status = status self.code = code @dataclass(frozen=True) class MintedKey: """A key that was made. `key` is the key itself, shown nowhere else: keep it. It is left out of the repr, so a log line that prints this object never carries it.""" key: str = field(repr=False) id: str store: str level: str label: str def mint(store: str, *, label: str, level: str = "read", email: str | None = None, password: str | None = None, token: str | None = None, open_with: str | None = None, api: str = DEFAULT_API, timeout: float = 60.0) -> MintedKey: """Make a key for `store`, listed in the dashboard as `label`, able to do what `level` names. See the module documentation for signing in.""" open_with = open_with.strip() if open_with else None token = token.strip() if token else None _check(store, label, level, email, password, token, open_with) account = _Account(api, timeout) master: bytes | None = None if token: account.token = token customer_id = account.who_am_i() else: assert email is not None and password is not None master = _master_key(password, email) customer_id = account.sign_in(email, _login_proof(master)) try: dek = _open_store(account, store, customer_id, open_with, master) key = _new_store_key() wrapped = _make_slot(dek, key, customer_id) made = account.add_key(store, _key_hash(key), wrapped, level, label.strip()) finally: account.sign_out() return MintedKey(key=key, id=str(made.get("id", "")), store=str(made.get("store", store)), level=str(made.get("level", level)), label=label.strip()) def _check(store: str, label: str, level: str, email: str | None, password: str | None, token: str | None, open_with: str | None) -> None: if not store: raise ValueError("store is required: the name of the store the key opens") if not label or not label.strip(): raise ValueError("label is required: the name the key is listed under in the dashboard") if level not in LEVELS: raise ValueError(f"level must be one of {', '.join(LEVELS)}") if email and token: raise ValueError("sign in one way: email and password, or a token, not both") if not email and not token: raise ValueError("sign in with email and password, or with an account token") if email and not password: raise ValueError("a password is needed with the email") if token and not open_with: raise ValueError("an account token signs in, but opens no store: give a key " "that opens it with open_with") if token and not token.startswith(_TOKEN_PREFIX): raise ValueError("that is not an account token: an account token begins dke_a_") if open_with and not open_with.startswith(_STORE_KEY_PREFIX) \ and _parse_customer_key(open_with) is None: raise ValueError("the key that opens the store must be one of its store keys " "(dke_s_...) or your customer key (dke_c_...)") # ── the format ───────────────────────────────────────────────────────────── def _hkdf(ikm: bytes, salt: bytes, info: str) -> bytes: """HKDF-SHA256 to 32 bytes (RFC 5869). An empty salt is a hash-length of zeros, as the RFC says.""" prk = hmac.new(salt or bytes(32), ikm, hashlib.sha256).digest() return hmac.new(prk, info.encode("utf-8") + b"\x01", hashlib.sha256).digest() def _seal(kek: bytes, plain: bytes) -> str: """base64(nonce || ciphertext || tag).""" nonce = secrets.token_bytes(_NONCE_LEN) return base64.b64encode(nonce + AESGCM(kek).encrypt(nonce, plain, None)).decode("ascii") def _open(kek: bytes, wrapped: str) -> bytes | None: """The reverse of `_seal`, or None for any wrap `kek` did not make.""" try: blob = base64.b64decode(wrapped, validate=True) except ValueError: return None if len(blob) < _NONCE_LEN + _TAG_LEN: return None try: plain = AESGCM(kek).decrypt(blob[:_NONCE_LEN], blob[_NONCE_LEN:], None) except InvalidTag: return None return plain if len(plain) == _KEY_LEN else None def _key_hash(store_key: str) -> str: """What the account service knows a store key by.""" return hashlib.sha256(store_key.encode("utf-8")).hexdigest() def _new_store_key() -> str: raw = secrets.token_bytes(_KEY_LEN) return _STORE_KEY_PREFIX + base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") def _slot_kek(store_key: str, customer_id: str) -> bytes: return _hkdf(store_key.encode("utf-8"), customer_id.encode("utf-8"), KEY_FORMAT) def _make_slot(dek: bytes, store_key: str, customer_id: str) -> str: return _seal(_slot_kek(store_key, customer_id), dek) def _open_slot(wrapped: str, store_key: str, customer_id: str) -> bytes | None: return _open(_slot_kek(store_key, customer_id), wrapped) def _open_customer_seal(sealed: str, customer_key: bytes, customer_id: str) -> bytes | None: return _open(_hkdf(customer_key, customer_id.encode("utf-8"), _CUSTOMER_SEAL_INFO), sealed) def _norm_email(email: str) -> str: """The address as the key derivation sees it: JavaScript's trim, then its lower case. A different address is a different master key, which the account reads as a wrong password.""" return email.strip(_JS_SPACE).lower() def _master_key(password: str, email: str) -> bytes: return hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), _norm_email(email).encode("utf-8"), _PBKDF2_ITERATIONS, _KEY_LEN) def _login_proof(master: bytes) -> str: return _hkdf(master, b"", _LOGIN_PROOF_INFO).hex() def _unwrap_customer_key(master: bytes, wrapped: str) -> bytes | None: return _open(_hkdf(master, b"", _CUSTOMER_KEY_INFO), wrapped) def _parse_customer_key(text: str) -> bytes | None: """A saved customer key back to its 32 bytes, in either spelling the dashboard has shown. A body that decodes but is not the one the key encodes to is refused, rather than read as a neighbouring key.""" m = _CUSTOMER_KEY_TEXT.fullmatch(text) if not m: return None key = base64.urlsafe_b64decode(m.group(1) + "=") if len(key) != _KEY_LEN or base64.urlsafe_b64encode(key).decode().rstrip("=") != m.group(1): return None return key # ── opening the store ────────────────────────────────────────────────────── def _open_store(account: _Account, store: str, customer_id: str, open_with: str | None, master: bytes | None) -> bytes: if open_with and open_with.startswith(_STORE_KEY_PREFIX): try: wrapped = account.slot(store, _key_hash(open_with)) except MintError as refused: if refused.status == 404: raise MintError(f"the key you gave does not open the store {store!r}: check that " "the store has that name and the key is one of its current keys", status=404) from None raise MintError(f"could not open the store {store!r}: {refused}", status=refused.status, code=refused.code) from None dek = _open_slot(wrapped, open_with, customer_id) if dek is None: raise MintError(f"the key you gave does not open the store {store!r}") return dek if open_with: customer_key = _parse_customer_key(open_with) else: assert master is not None held = account.customer_key() wrap = held.get("customer_key_wrap") if not isinstance(wrap, str): raise MintError("your account has no customer key yet: sign in to the dashboard " "once, and one is made for you") customer_key = _unwrap_customer_key(master, wrap) if customer_key is None: if held.get("customer_key_stale") is True: raise MintError("your password was changed since your customer key was put " "under it, so the password no longer opens the key: sign in to " "the dashboard to bring it back, or give the customer key you " "saved with open_with") raise MintError("your password does not open your customer key") assert customer_key is not None try: sealed = account.customer_seal(store) except MintError as refused: if refused.code == "no_customer_slot": raise MintError(f"the store {store!r} is not added to your customer key: open it " "with one of its store keys (open_with), or add it to your " "customer key in the dashboard", status=404, code=refused.code) from None if refused.status == 404: raise MintError(f"there is no store named {store!r}", status=404) from None raise MintError(f"could not open the store {store!r}: {refused}", status=refused.status, code=refused.code) from None dek = _open_customer_seal(sealed, customer_key, customer_id) if dek is None: raise MintError(f"your customer key does not open the store {store!r}: it is sealed " "to an earlier customer key. Open it with one of its store keys " "(open_with), or add it to your customer key in the dashboard") return dek # ── the account service ──────────────────────────────────────────────────── class _Account: """A signed-in client of the account service. It holds a session or a token, never a key that opens a store: every call sends hashes and wraps.""" def __init__(self, api: str, timeout: float) -> None: self.base = api.rstrip("/") self.timeout = timeout self.session: str | None = None self.token: str | None = None self._opener = urllib.request.build_opener(_NoRedirect) def _call(self, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[dict[str, Any], list[str]]: """One request. Answers the JSON body and the Set-Cookie headers; any answer outside 2xx raises MintError carrying the service's words.""" headers = {"Accept": "application/json", "User-Agent": f"dkepy/{__version__}"} data = None if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" if self.token: headers["Authorization"] = f"Bearer {self.token}" if self.session: headers["Cookie"] = f"{_SESSION_COOKIE}={self.session}" request = urllib.request.Request(self.base + path, data=data, headers=headers, method=method) try: with self._opener.open(request, timeout=self.timeout) as response: raw = response.read(1 << 20) cookies = response.headers.get_all("Set-Cookie") or [] except urllib.error.HTTPError as refused: try: if 300 <= refused.code < 400: raise MintError(f"the account service at {self.base} answered with a " "redirect, which is not followed: check the address", status=refused.code) from None answer = _json(refused.read(1 << 20)) finally: refused.close() raise MintError(str(answer.get("error") or f"the account service answered " f"{refused.code}"), status=refused.code, code=answer.get("code")) from None except (urllib.error.URLError, OSError) as unreachable: raise MintError(f"could not reach the account service: {unreachable}") from None return _json(raw), cookies def sign_in(self, email: str, proof: str) -> str: try: me, cookies = self._call("POST", "/v1/login", {"email": _norm_email(email), "login_proof": proof}) except MintError as refused: if refused.status == 401: raise MintError("could not sign in: wrong email or password", status=401) from None raise MintError(f"could not sign in: {refused}", status=refused.status, code=refused.code) from None for cookie in cookies: name, _, rest = cookie.partition("=") if name.strip() == _SESSION_COOKIE: self.session = rest.split(";", 1)[0] or None customer_id = me.get("customer_id") if not self.session or not isinstance(customer_id, str) or not customer_id: raise MintError("the account service signed in without starting a session") return customer_id def sign_out(self) -> None: """End the session sign_in started. A session left behind would end on its own; this only makes it end sooner.""" if not self.session: return try: self._call("POST", "/v1/logout") except MintError: pass self.session = None def who_am_i(self) -> str: try: who, _ = self._call("GET", "/v1/token") except MintError as refused: raise MintError(f"could not sign in with the account token: {refused}", status=refused.status, code=refused.code) from None customer_id = who.get("customer_id") if not isinstance(customer_id, str) or not customer_id: raise MintError("the account service did not say whose token this is") return customer_id def customer_key(self) -> dict[str, Any]: try: held, _ = self._call("GET", "/v1/customer-key") except MintError as refused: raise MintError(f"could not read your customer key: {refused}", status=refused.status, code=refused.code) from None return held def slot(self, store: str, key_hash: str) -> str: answer, _ = self._call("POST", "/v1/keys/slot", {"store": store, "key_hash": key_hash}) wrapped = answer.get("wrapped_slot") if not isinstance(wrapped, str) or not wrapped: raise MintError("the account service answered with no slot") return wrapped def customer_seal(self, store: str) -> str: answer, _ = self._call( "GET", "/v1/stores/" + urllib.parse.quote(store, safe="") + "/customer-slot") sealed = answer.get("customer_slot") if not isinstance(sealed, str) or not sealed: raise MintError("the account service answered with no seal") return sealed def add_key(self, store: str, key_hash: str, wrapped: str, level: str, label: str) -> dict[str, Any]: """`create` is False: this adds keys to stores that exist, and never makes one.""" try: made, _ = self._call("POST", "/v1/keys/minted", { "store": store, "key_hash": key_hash, "wrapped_slot": wrapped, "slot_format": KEY_FORMAT, "level": level, "label": label, "create": False, }) except MintError as refused: if refused.status == 404: raise MintError(f"there is no store named {store!r}", status=404) from None raise MintError(f"the key was not made: {refused}", status=refused.status, code=refused.code) from None return made def _json(raw: bytes) -> dict[str, Any]: try: value = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return {} return value if isinstance(value, dict) else {}