.gitignore

@@ -5,4 +5,4 @@ dist/
 build/
 .venv/
 .pytest_cache/
-.claude/
+CLAUDE.md

README.md

@@ -13,27 +13,25 @@ authorization system and the key-document schema; the crypto primitives live in
 ## Install
 
 ```
-envelope_authorizer @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.2
+envelope_authorizer @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.0
 ```
 
 Direct:
 
 ```bash
-pip install "envelope_authorizer @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.2"
+pip install "envelope_authorizer @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.0"
 ```
 
 The base install uses a local JSON file for storage (stdlib only). For shared
 dev→server storage, install the mongo extra:
 
 ```bash
-pip install "envelope_authorizer[mongo] @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.2"
+pip install "envelope_authorizer[mongo] @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_authorizer.git@v0.1.0"
 ```
 
 Installing pulls `envelope_crypto` (and `mongo` with the extra). After install,
 the `authorizer` command is on your PATH; `python -m envelope_authorizer` also works.
 
-Drop the `@v0.1.2` suffix from the line above to install the latest unpinned.
-
 ## Trust model (read this)
 
 There is one shared **AES data-encryption key (DEK)** per project. Each key doc
@@ -49,39 +47,12 @@ Each key carries an **encrypted capability flag** — `meta.authorizer` =
 - **Servers** are authorized with `allowed: False` → they can boot and use the DEK
   (decrypt project data) but **cannot authorize anything else**.
 
-The flag is GCM-sealed under the DEK on purpose. What that buys, stated honestly:
+The flag is GCM-sealed under the DEK on purpose: a server **cannot flip its own
-
+flag in the database to escalate**, because forging a valid flag requires already
-- **Without the DEK, the flag is inert.** A stolen database at rest cannot be
+holding the DEK, and editing the stored ciphertext breaks the auth tag. The
-  escalated — the flags are ciphertext, unforgeable and unflippable without the key
+guarantee is precisely: *you cannot grant yourself authority without already
-  (editing the stored ciphertext breaks the GCM auth tag). This stops passive DB
+having it.* (A legitimate authorizer can of course mint new flags — that is its
-  tampering.
+job. This stops passive DB tampering, not an authorizer.)
-- **It does not defend against a party that already holds the DEK and can write
-  storage.** The flag is sealed under the *shared* DEK and is **not bound to key
-  identity**, so such a party can copy a valid sealed `True` flag from an
-  authorizer's doc onto its own and self-grant authority. This is **out of scope by
-  design**: the trust model assumes DEK-holders are trusted. A DEK-holder already
-  has full data access (DEK + write access is game over), so self-granting authority
-  exposes no additional data. We deliberately do not add per-key signing to prevent
-  it — the residual is handled operationally (below).
-
-So the precise boundary is: **the capability flag is unforgeable without the DEK; it
-is not a defense against a malicious DEK-holder.** (A legitimate authorizer can of
-course mint new flags — that is its job.)
-
-### Operational guidance (the residual control)
-
-Because the DEK-holder case above is **not** prevented cryptographically in the
-shared-DEK model, the intended mitigation is **detection, not crypto**. Consumers
-should periodically audit the authorization state in storage:
-
-- enumerate every key doc's capability flag (`list` shows authorizer status),
-- compare the set of `allowed: True` keys against a known-good list of expected
-  authorizers,
-- alert on any unexpected authorizer.
-
-An unexpected `allowed: True` appearing in storage is the signal that a DEK-holder
-self-granted — catch it by review, then revoke and re-authorize from a trusted
-machine.
 
 `revoke` is **bookkeeping, not rotation**: it deletes the record so the key can no
 longer unwrap at boot, but it does not rotate the DEK or scrub it from a machine
@@ -224,4 +195,5 @@ Owned by this lib (not `envelope_crypto`):
 
 ## Versioning
 
-Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs.
+Tagged `vX.Y.Z`. Pin the tag. `envelope_crypto` is pinned at `v0.1.0` in
+`pyproject.toml`; to change it, edit the pin and re-test.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "envelope_authorizer"
-version = "0.1.2"
+version = "0.1.0"
 description = "CLI key-authorization manager for envelope_crypto"
 requires-python = ">=3.10"
 dependencies = [

src/envelope_authorizer/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.2"
+__version__ = "0.1.0"

src/envelope_authorizer/cli.py

@@ -9,8 +9,6 @@ command loads config and resolves a storage backend first. expected failures
 import argparse
 import sys
 
-from cryptography.exceptions import InvalidTag
-
 from . import __version__
 from .config import ConfigError, load_config
 from .commands import CommandError, authorize, config_init, init, list_keys, revoke, verify
@@ -72,9 +70,7 @@ def main() -> int:
             try:
                 config_init.run(None, None, args)
                 return 0
-            except (CommandError, OSError) as error:
+            except CommandError as error:
-                # config_init writes a file (cwd may be read-only, full, or gone) — an
-                # OSError must print a clean [✘] line, not a raw traceback
                 return _fail(str(error))
         parser.parse_args(["config", "--help"])
         return 0
@@ -92,12 +88,7 @@ def main() -> int:
         storage = resolve(config)
         handlers[args.cmd](config, storage, args)
         return 0
-    except InvalidTag:
+    except (ConfigError, CommandError, RuntimeError, ValueError, FileNotFoundError) as error:
-        return _fail("capability flag failed authentication — tampered or wrong DEK")
-    except (ConfigError, CommandError, RuntimeError, ValueError, OSError, KeyError, TypeError) as error:
-        # OSError covers the FileNotFoundError/PermissionError/IsADirectoryError family;
-        # KeyError/TypeError cover a structurally-malformed flag/doc (unguarded indexing
-        # of ['iv']/['meta']['authorizer']/['key']) — all print a clean [✘] line, not a traceback
         return _fail(str(error))
 
 

src/envelope_authorizer/commands/__init__.py

@@ -22,11 +22,7 @@ def make_flag(crypto: EnvelopeCrypto, allowed: bool) -> dict:
 
 
 def read_flag(crypto: EnvelopeCrypto, blob: dict) -> bool:
-    """decrypt a capability flag; return the `allowed` bool
+    """decrypt a capability flag; return the `allowed` bool"""
-
-    fails closed: a non-dict / unexpected plaintext reads as not-allowed rather than
-    raising — a privilege gate must default to deny on a malformed flag.
-    """
     data = crypto.decrypt_data(blob)
     return bool(data.get("allowed", False)) if isinstance(data, dict) else False
 

src/envelope_authorizer/commands/init.py

@@ -11,14 +11,7 @@ from . import CommandError, build_doc, find_by_friendly, make_flag
 
 
 def run(config, storage, args) -> None:
-    """initialize the key system on this machine as the first authorizer
+    """initialize the key system on this machine as the first authorizer"""
-
-    the already-initialized / duplicate-friendly checks are non-atomic (a check-then-act
-    TOCTOU under two concurrent CLIs), but this is a one-shot human admin tool and `save`
-    upserts by `_id`, so key material can never collide — the worst case is a cosmetic
-    double-init under a race, which carries no security consequence in the trusted-
-    DEK-holder threat model. left non-atomic by design.
-    """
     if storage.get_all():
         raise CommandError(
             "already initialized; use `authorizer list` to see existing keys"

src/envelope_authorizer/commands/list_keys.py

@@ -11,12 +11,7 @@ from . import boot_local, read_flag
 
 
 def _can_authorize(crypto, doc) -> str:
-    """decrypted authority of a doc as Yes/No, or `?` if not readable here
+    """decrypted authority of a doc as Yes/No, or `?` if not readable here"""
-
-    `?` means the flag could not be read for ANY reason — the local key can't unwrap it,
-    or the doc is missing/malformed — so the table always renders rather than crashing on
-    one bad row. it is not specifically a corruption signal.
-    """
     try:
         return "Yes" if read_flag(crypto, doc["meta"]["authorizer"]) else "No"
     except Exception:
@@ -24,16 +19,11 @@ def _can_authorize(crypto, doc) -> str:
 
 
 def _created(doc) -> str:
-    """format created_at as a UTC timestamp string, or '-' if absent/unparseable"""
+    """format created_at as a UTC timestamp string, or '-' if absent"""
     raw = doc.get("meta", {}).get("created_at")
     if not raw:
         return "-"
-    try:
+    return datetime.fromtimestamp(int(raw), tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
-        return datetime.fromtimestamp(int(raw), tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
-    except (TypeError, ValueError, OverflowError, OSError):
-        # one hand-edited/legacy doc with a bad created_at shouldn't abort the
-        # whole table render
-        return "-"
 
 
 def run(config, storage, args) -> None:

src/envelope_authorizer/commands/revoke.py

@@ -18,18 +18,11 @@ _WARNING = (
 
 
 def _find_by_fingerprint(storage, prefix: str):
-    """return the single doc whose `_id` starts with the given prefix, or None
+    """return the doc whose `_id` starts with the given prefix, or None"""
-
+    for doc in storage.get_all():
-    rejects an empty prefix (which would match every key) and an ambiguous prefix
+        if doc.get("_id", "").startswith(prefix):
-    that matches more than one key, rather than silently revoking the first match.
+            return doc
-    """
+    return None
-    if not prefix:
-        raise CommandError("fingerprint prefix must not be empty")
-    matches = [doc for doc in storage.get_all() if doc.get("_id", "").startswith(prefix)]
-    if len(matches) > 1:
-        ids = ", ".join(d["_id"][:16] for d in matches)
-        raise CommandError(f"fingerprint prefix '{prefix}' is ambiguous; matches: {ids}")
-    return matches[0] if matches else None
 
 
 def run(config, storage, args) -> None:

src/envelope_authorizer/storage/json_store.py

@@ -8,7 +8,6 @@ setups; for dev->server flows across machines, use the mongo backend.
 
 import json
 import os
-import tempfile
 from pathlib import Path
 from typing import List, Optional
 
@@ -35,36 +34,13 @@ class JsonStore(StorageBackend):
         return data
 
     def _write(self, docs: List[dict]) -> None:
-        """atomically write the docs list (unique temp file then replace)
+        """atomically write the docs list (temp file then replace)"""
-
-        the temp name is unique per write (tempfile.mkstemp in the same dir) so two
-        concurrent writers can't clobber a shared `.tmp`; os.replace is atomic on the
-        same filesystem. the temp is cleaned up if the write fails before replace.
-        """
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        fd, tmp = tempfile.mkstemp(
+        tmp = self.path.with_suffix(self.path.suffix + ".tmp")
-            dir=self.path.parent, prefix=self.path.name + ".", suffix=".tmp"
+        with open(tmp, "w", encoding="utf-8") as handle:
-        )
+            json.dump(docs, handle, indent=2)
-        wrapped = False
+            handle.write("\n")
-        try:
+        os.replace(tmp, self.path)
-            with os.fdopen(fd, "w", encoding="utf-8") as handle:
-                wrapped = True  # fdopen took ownership of fd; its close() handles it
-                json.dump(docs, handle, indent=2)
-                handle.write("\n")
-            os.replace(tmp, self.path)
-        except BaseException:
-            if not wrapped:
-                # fdopen raised before taking ownership — close the raw fd ourselves so
-                # it isn't leaked (the `with` only closes once fdopen returns a file object)
-                try:
-                    os.close(fd)
-                except OSError:
-                    pass
-            try:
-                os.unlink(tmp)
-            except OSError:
-                pass
-            raise
 
     def get(self, fingerprint: str) -> Optional[dict]:
         """return the doc whose `_id` matches, or None"""

src/envelope_authorizer/storage/mongo_store.py

@@ -6,14 +6,6 @@ self-contained per call. no module-level client and no shared event loop, so it
 never collides with a running loop. a per-call connect is an accepted tradeoff
 for a one-shot admin CLI. requires envelope_authorizer[mongo]; without it every
 method raises a clear RuntimeError.
-
-fail-loud: operations go through the mongo lib's RAW collection escape hatch
-(`db.collection(name)`, the motor collection — which raises) rather than the
-swallow-and-return-default wrapped methods. this is an auth backend; conflating
-a backend error with "no document / not initialized" in the thing that gates
-authority is the worst place for the swallow anti-pattern, so a driver error
-propagates and a no-op upsert raises instead of silently reporting success.
-`close()` is synchronous on the mongo lib (motor's close is sync) — not awaited.
 """
 
 import asyncio
@@ -44,38 +36,33 @@ class MongoStore(StorageBackend):
     async def _get(self, fingerprint: str) -> Optional[dict]:
         db = Mongo(self.uri, self.database)
         try:
-            return await db.collection(self.collection).find_one({"_id": fingerprint})
+            return await db.get_document(self.collection, {"_id": fingerprint})
         finally:
-            db.close()
+            await db.close()
 
     async def _get_all(self) -> List[dict]:
         db = Mongo(self.uri, self.database)
         try:
-            cursor = db.collection(self.collection).find({})
+            return await db.get_documents(self.collection, {})
-            return await cursor.to_list(length=None)
         finally:
-            db.close()
+            await db.close()
 
     async def _save(self, doc: dict) -> None:
         db = Mongo(self.uri, self.database)
         try:
-            result = await db.collection(self.collection).replace_one(
+            await db.update_document(
-                {"_id": doc["_id"]}, doc, upsert=True
+                self.collection, {"_id": doc["_id"]}, doc, do_upsert=True
             )
-            if not (result.matched_count or result.upserted_id):
-                raise RuntimeError(
-                    f"mongo upsert affected no document for _id={doc['_id']!r}"
-                )
         finally:
-            db.close()
+            await db.close()
 
     async def _delete(self, fingerprint: str) -> bool:
         db = Mongo(self.uri, self.database)
         try:
-            result = await db.collection(self.collection).delete_one({"_id": fingerprint})
+            removed = await db.delete_document(self.collection, {"_id": fingerprint})
-            return result.deleted_count > 0
+            return bool(removed)
         finally:
-            db.close()
+            await db.close()
 
     def get(self, fingerprint: str) -> Optional[dict]:
         """return the key doc with this `_id`, or None"""