Skip to content

zettelkasten.synapse.sharing

zettelkasten.synapse.sharing

Encryption-first sharing for the committed .synapse link overlay.

The THIRD sharing bundle, alongside :mod:memory.sharing and :mod:zettelkasten.sharing, but for cross-store SYNAPSE EDGES — the bipartite memory <-> zk links in .synapse/links/links.json. Because an edge names ids from BOTH stores, it is shared ONLY to recipients who can already open BOTH endpoints (the intersection of the two stores' per-unit recipient maps); see :func:edge_recipients. This is the whole id-secrecy property.

Crypto and identity are REUSED from :mod:memory.sharing unchanged — there is a single firm identity registry (.memory/identities.yaml) and one local key (~/.angelo/identity.yaml); this package does NOT fork them. Only the edge-policy and bundle layers are synapse-specific.

Typical producer flow::

from memory import sharing as mem_sharing
from zettelkasten.synapse import sharing as syn_sharing
registry = mem_sharing.load_registry()
result = syn_sharing.build_bundle("/path/to/bundle-repo", registry=registry)

Typical consumer flow::

from zettelkasten.synapse import sharing as syn_sharing
syn_sharing.decrypt_bundle(
    "/path/to/fetched-bundle-repo",
    "/path/to/federation-cache/peer",   # writes <peer>/.synapse/links/links.json
)  # uses ~/.angelo/identity.yaml

IdentityRegistry dataclass

An id -> RegistryEntry map of publishable (public-key) identities.

Source code in memory/sharing/identity.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@dataclass(frozen=True)
class IdentityRegistry:
    """An ``id -> RegistryEntry`` map of publishable (public-key) identities."""

    entries: Mapping[str, RegistryEntry]

    def member_ids(self) -> set[str]:
        """All registered recipient ids (used to expand a ``firm`` scope)."""
        return set(self.entries)

    def public_key_for(self, recipient_id: str) -> str | None:
        """The public key for ``recipient_id``, or ``None`` if unregistered."""
        entry = self.entries.get(recipient_id)
        return entry.public_key if entry else None

    def signing_public_key_for(self, recipient_id: str) -> str | None:
        """The Ed25519 signing public key for ``recipient_id``.

        Returns ``None`` when the id is unregistered *or* when the registered
        entry predates signing (legacy entry with no ``signing_public_key``).
        """
        entry = self.entries.get(recipient_id)
        return entry.signing_public_key if entry else None

    def public_keys_for(self, recipient_ids: Iterable[str]) -> list[str]:
        """Resolve ids to public keys, silently dropping unregistered ids."""
        keys: list[str] = []
        for rid in recipient_ids:
            pk = self.public_key_for(rid)
            if pk:
                keys.append(pk)
        return keys

    def with_entry(self, entry: RegistryEntry) -> "IdentityRegistry":
        """A copy of this registry with ``entry`` added/overwritten (immutably)."""
        merged = dict(self.entries)
        merged[entry.id] = entry
        return IdentityRegistry(merged)

member_ids 

member_ids() -> set[str]

All registered recipient ids (used to expand a firm scope).

Source code in memory/sharing/identity.py 
150
151
152
def member_ids(self) -> set[str]:
    """All registered recipient ids (used to expand a ``firm`` scope)."""
    return set(self.entries)

public_key_for 

public_key_for(recipient_id: str) -> str | None

The public key for recipient_id, or None if unregistered.

Source code in memory/sharing/identity.py 
154
155
156
157
def public_key_for(self, recipient_id: str) -> str | None:
    """The public key for ``recipient_id``, or ``None`` if unregistered."""
    entry = self.entries.get(recipient_id)
    return entry.public_key if entry else None

signing_public_key_for 

signing_public_key_for(recipient_id: str) -> str | None

The Ed25519 signing public key for recipient_id.

Returns None when the id is unregistered or when the registered entry predates signing (legacy entry with no signing_public_key).

Source code in memory/sharing/identity.py 
159
160
161
162
163
164
165
166
def signing_public_key_for(self, recipient_id: str) -> str | None:
    """The Ed25519 signing public key for ``recipient_id``.

    Returns ``None`` when the id is unregistered *or* when the registered
    entry predates signing (legacy entry with no ``signing_public_key``).
    """
    entry = self.entries.get(recipient_id)
    return entry.signing_public_key if entry else None

public_keys_for 

public_keys_for(recipient_ids: Iterable[str]) -> list[str]

Resolve ids to public keys, silently dropping unregistered ids.

Source code in memory/sharing/identity.py 
168
169
170
171
172
173
174
175
def public_keys_for(self, recipient_ids: Iterable[str]) -> list[str]:
    """Resolve ids to public keys, silently dropping unregistered ids."""
    keys: list[str] = []
    for rid in recipient_ids:
        pk = self.public_key_for(rid)
        if pk:
            keys.append(pk)
    return keys

with_entry 

with_entry(entry: RegistryEntry) -> 'IdentityRegistry'

A copy of this registry with entry added/overwritten (immutably).

Source code in memory/sharing/identity.py 
177
178
179
180
181
def with_entry(self, entry: RegistryEntry) -> "IdentityRegistry":
    """A copy of this registry with ``entry`` added/overwritten (immutably)."""
    merged = dict(self.entries)
    merged[entry.id] = entry
    return IdentityRegistry(merged)

LocalIdentity dataclass

This machine's identity: a logical id plus its keypair(s).

Persisted (with the private keys) to ~/.angelo/identity.yaml. Carries both the age X25519 encryption keypair and the Ed25519 signing keypair; the signing fields are optional (default None) so a legacy identity file with no signing key still loads.

Source code in memory/sharing/identity.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@dataclass(frozen=True)
class LocalIdentity:
    """This machine's identity: a logical id plus its keypair(s).

    Persisted (with the private keys) to ``~/.angelo/identity.yaml``. Carries both
    the ``age`` X25519 encryption keypair and the Ed25519 signing keypair; the
    signing fields are optional (default ``None``) so a legacy identity file with
    no signing key still loads.
    """

    id: str
    private_key: str
    public_key: str
    display_name: str | None = None
    github: str | None = None
    signing_public_key: str | None = None
    signing_private_key: str | None = None

    def to_pyrage_identity(self) -> x25519.Identity:
        """Parse the private key into a ``pyrage`` identity for decryption."""
        return x25519.Identity.from_str(self.private_key)

    def to_registry_entry(self) -> RegistryEntry:
        """The public-only :class:`RegistryEntry` for publishing to the registry."""
        return RegistryEntry(
            id=self.id,
            public_key=self.public_key,
            display_name=self.display_name,
            github=self.github,
            signing_public_key=self.signing_public_key,
        )

to_pyrage_identity 

to_pyrage_identity() -> Identity

Parse the private key into a pyrage identity for decryption.

Source code in memory/sharing/identity.py 
129
130
131
def to_pyrage_identity(self) -> x25519.Identity:
    """Parse the private key into a ``pyrage`` identity for decryption."""
    return x25519.Identity.from_str(self.private_key)

to_registry_entry 

to_registry_entry() -> RegistryEntry

The public-only :class:RegistryEntry for publishing to the registry.

Source code in memory/sharing/identity.py 
133
134
135
136
137
138
139
140
141
def to_registry_entry(self) -> RegistryEntry:
    """The public-only :class:`RegistryEntry` for publishing to the registry."""
    return RegistryEntry(
        id=self.id,
        public_key=self.public_key,
        display_name=self.display_name,
        github=self.github,
        signing_public_key=self.signing_public_key,
    )

RegistryEntry dataclass

One published identity: a logical id bound to a public key.

display_name and github are optional and used only for attribution / discovery — never for access control (the public key is the anchor). signing_public_key is the identity's Ed25519 signing public key (base64, public part only); optional so legacy registries without it still parse.

Source code in memory/sharing/identity.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@dataclass(frozen=True)
class RegistryEntry:
    """One published identity: a logical ``id`` bound to a public key.

    ``display_name`` and ``github`` are optional and used only for attribution /
    discovery — never for access control (the public key is the anchor).
    ``signing_public_key`` is the identity's Ed25519 *signing* public key (base64,
    public part only); optional so legacy registries without it still parse.
    """

    id: str
    public_key: str
    display_name: str | None = None
    github: str | None = None
    signing_public_key: str | None = None

BuildResult dataclass

Summary of a :func:build_bundle run (returned in memory, never written).

Source code in zettelkasten/synapse/sharing/bundle.py 
370
371
372
373
374
375
376
377
378
@dataclass(frozen=True)
class BuildResult:
    """Summary of a :func:`build_bundle` run (returned in memory, never written)."""

    manifest: BundleManifest
    output_dir: Path
    edge_unit_count: int
    recipient_ids: tuple[str, ...]
    skipped_unregistered_ids: tuple[str, ...]

BundleManifest dataclass

The cleartext manifest.json at the bundle root.

Source code in zettelkasten/synapse/sharing/bundle.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
@dataclass(frozen=True)
class BundleManifest:
    """The cleartext ``manifest.json`` at the bundle root."""

    format: str
    format_version: str
    created_at: str
    units: tuple[BundleUnit, ...]
    producer_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        data: dict[str, Any] = {
            "format": self.format,
            "format_version": self.format_version,
            "created_at": self.created_at,
            "units": [u.to_dict() for u in self.units],
        }
        if self.producer_id:
            data["producer_id"] = self.producer_id
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "BundleManifest":
        raw_units = data.get("units") or []
        units = tuple(BundleUnit.from_dict(u) for u in raw_units if isinstance(u, dict))
        producer = data.get("producer_id")
        return cls(
            format=str(data.get("format") or ""),
            format_version=str(data.get("format_version") or ""),
            created_at=str(data.get("created_at") or ""),
            units=units,
            producer_id=str(producer) if producer else None,
        )

BundleUnit dataclass

A pointer to one encrypted edge unit (kind + relative path).

Source code in zettelkasten/synapse/sharing/bundle.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
@dataclass(frozen=True)
class BundleUnit:
    """A pointer to one encrypted edge unit (``kind`` + relative path)."""

    kind: str
    path: str

    def to_dict(self) -> dict[str, str]:
        return {"kind": self.kind, "path": self.path}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "BundleUnit":
        return cls(kind=str(data.get("kind") or ""), path=str(data.get("path") or ""))

DecryptResult dataclass

Summary of a :func:decrypt_bundle run.

Source code in zettelkasten/synapse/sharing/bundle.py 
381
382
383
384
385
386
387
@dataclass(frozen=True)
class DecryptResult:
    """Summary of a :func:`decrypt_bundle` run."""

    output_dir: Path
    edge_count: int
    total_units: int

StalePeerPolicyError

Bases: RuntimeError

Raised by :func:build_bundle under the OPT-IN fail-closed staleness TTL.

When max_policy_age_hours is configured, a synapse publish REFUSES to ship cross-producer edges whose peer recipient-index policy is missing or older than the TTL — a "publish-requires-recent-sync" gate — rather than emitting a possibly over-permissive (stale) edge set. It is raised BEFORE any bundle output is written, so an existing bundle is left intact on failure.

stale_peers is a list of (peer_id, decrypted_at_iso_or_None) for programmatic handling; None means the peer's freshness stamp was missing/unparseable.

Source code in zettelkasten/synapse/sharing/bundle.py 
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
class StalePeerPolicyError(RuntimeError):
    """Raised by :func:`build_bundle` under the OPT-IN fail-closed staleness TTL.

    When ``max_policy_age_hours`` is configured, a synapse publish REFUSES to ship
    cross-producer edges whose peer recipient-index policy is missing or older than
    the TTL — a "publish-requires-recent-sync" gate — rather than emitting a
    possibly over-permissive (stale) edge set. It is raised BEFORE any bundle
    output is written, so an existing bundle is left intact on failure.

    ``stale_peers`` is a list of ``(peer_id, decrypted_at_iso_or_None)`` for
    programmatic handling; ``None`` means the peer's freshness stamp was
    missing/unparseable.
    """

    def __init__(self, message: str, stale_peers: list[tuple[str, str | None]]):
        super().__init__(message)
        self.stale_peers = stale_peers

SynapseSharingConfig dataclass

Parsed synapse sharing plumbing (no per-edge scopes — those are derived).

  • bundle_repo — producer-owned repo path the publisher writes into.
  • producer_id — optional attribution id stamped in the bundle manifest.
  • bundle_subscriptions — consumer subscription dicts (each an id plus a local path or a git remote), mirroring the memory/zk shape.
  • recipient_index_max_age_hours — OPT-IN fail-closed staleness TTL (default None = disabled). When set to a positive number, synapse publish REFUSES to publish if any cross-producer edge relies on a peer recipient-index policy that is missing or older than this many hours (a "publish-requires-recent-sync" gate). Left unset, publish keeps its default forward-only behaviour and only emits a non-fatal staleness warning.
Source code in zettelkasten/synapse/sharing/config.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class SynapseSharingConfig:
    """Parsed synapse sharing plumbing (no per-edge scopes — those are derived).

    * ``bundle_repo`` — producer-owned repo path the publisher writes into.
    * ``producer_id`` — optional attribution id stamped in the bundle manifest.
    * ``bundle_subscriptions`` — consumer subscription dicts (each an ``id`` plus
      a local ``path`` or a git ``remote``), mirroring the memory/zk shape.
    * ``recipient_index_max_age_hours`` — OPT-IN fail-closed staleness TTL (default
      ``None`` = disabled). When set to a positive number, ``synapse publish``
      REFUSES to publish if any cross-producer edge relies on a peer
      recipient-index policy that is missing or older than this many hours (a
      "publish-requires-recent-sync" gate). Left unset, publish keeps its default
      forward-only behaviour and only emits a non-fatal staleness *warning*.
    """

    bundle_repo: str | None = None
    producer_id: str | None = None
    bundle_subscriptions: tuple[Mapping[str, Any], ...] = field(default_factory=tuple)
    recipient_index_max_age_hours: float | None = None

RecipientIndexBuildResult dataclass

Summary of a :func:build_recipient_index run.

Source code in zettelkasten/synapse/sharing/recipient_index.py 
215
216
217
218
219
220
221
222
223
224
225
226
@dataclass(frozen=True)
class RecipientIndexBuildResult:
    """Summary of a :func:`build_recipient_index` run."""

    output_dir: Path
    note_count: int
    recipient_ids: tuple[str, ...]
    skipped_unregistered_ids: tuple[str, ...]
    written: bool
    # True iff the export was published WITH an Ed25519 signature over the
    # canonical mapping bytes; False for an unsigned (no-signing-key) publish.
    signed: bool = False

RecipientIndexDecryptResult dataclass

Summary of a :func:decrypt_recipient_index run.

Source code in zettelkasten/synapse/sharing/recipient_index.py 
229
230
231
232
233
234
@dataclass(frozen=True)
class RecipientIndexDecryptResult:
    """Summary of a :func:`decrypt_recipient_index` run."""

    output_dir: Path
    note_count: int

decrypt 

decrypt(ciphertext: bytes, private_key: PrivateKey) -> bytes

Decrypt ciphertext with private_key.

private_key may be an AGE-SECRET-KEY-... string or a pre-parsed :class:pyrage.x25519.Identity. Raises :class:DecryptError if the ciphertext was not encrypted to this key.

Source code in memory/sharing/crypto.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
def decrypt(ciphertext: bytes, private_key: PrivateKey) -> bytes:
    """Decrypt ``ciphertext`` with ``private_key``.

    ``private_key`` may be an ``AGE-SECRET-KEY-...`` string or a pre-parsed
    :class:`pyrage.x25519.Identity`. Raises :class:`DecryptError` if the
    ciphertext was not encrypted to this key.
    """
    identity = (
        private_key
        if isinstance(private_key, x25519.Identity)
        else x25519.Identity.from_str(str(private_key))
    )
    return pyrage.decrypt(bytes(ciphertext), [identity])

encrypt 

encrypt(plaintext: bytes, recipient_pubkeys: list[str]) -> bytes

Encrypt plaintext to every recipient in recipient_pubkeys.

Each key is an age X25519 recipient string (age1...). Any recipient holding the matching private key can :func:decrypt the result. Returns the binary age ciphertext.

Raises :class:ValueError if no recipients are given (encrypting to nobody would produce an unopenable blob) or if a recipient string is malformed (surfaced as :class:pyrage.RecipientError).

Source code in memory/sharing/crypto.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def encrypt(plaintext: bytes, recipient_pubkeys: list[str]) -> bytes:
    """Encrypt ``plaintext`` to every recipient in ``recipient_pubkeys``.

    Each key is an ``age`` X25519 recipient string (``age1...``). Any recipient
    holding the matching private key can :func:`decrypt` the result. Returns the
    binary ``age`` ciphertext.

    Raises :class:`ValueError` if no recipients are given (encrypting to nobody
    would produce an unopenable blob) or if a recipient string is malformed
    (surfaced as :class:`pyrage.RecipientError`).
    """
    if not recipient_pubkeys:
        raise ValueError("at least one recipient public key is required to encrypt")
    recipients = [x25519.Recipient.from_str(pk) for pk in _dedupe(list(recipient_pubkeys))]
    return pyrage.encrypt(bytes(plaintext), recipients)

create_identity 

create_identity(identity_id: str, *, display_name: str | None = None, github: str | None = None, path: Path | str | None = None, overwrite: bool = False) -> LocalIdentity

Generate a keypair, build a :class:LocalIdentity, and persist it.

Refuses to clobber an existing identity file unless overwrite=True — a lost private key cannot be recovered, so overwriting is opt-in.

Source code in memory/sharing/identity.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def create_identity(
    identity_id: str,
    *,
    display_name: str | None = None,
    github: str | None = None,
    path: Path | str | None = None,
    overwrite: bool = False,
) -> LocalIdentity:
    """Generate a keypair, build a :class:`LocalIdentity`, and persist it.

    Refuses to clobber an existing identity file unless ``overwrite=True`` — a
    lost private key cannot be recovered, so overwriting is opt-in.
    """
    dest = Path(path) if path is not None else default_identity_path()
    if dest.exists() and not overwrite:
        raise FileExistsError(
            f"identity file already exists at {dest}; pass overwrite=True to replace it"
        )
    keypair = generate_keypair()
    identity = LocalIdentity(
        id=identity_id,
        private_key=keypair.private_key,
        public_key=keypair.public_key,
        display_name=display_name,
        github=github,
        signing_public_key=keypair.signing_public_key,
        signing_private_key=keypair.signing_private_key,
    )
    save_local_identity(identity, dest)
    return identity

default_identity_path 

default_identity_path() -> Path

Path of this machine's private identity file: ~/.angelo/identity.yaml.

Under HOME (never the workspace) so the private key cannot be committed.

Source code in memory/sharing/identity.py 
55
56
57
58
59
60
def default_identity_path() -> Path:
    """Path of this machine's private identity file: ``~/.angelo/identity.yaml``.

    Under HOME (never the workspace) so the private key cannot be committed.
    """
    return Path.home() / ".angelo" / "identity.yaml"

default_registry_path 

default_registry_path() -> Path

Default path of the shared identity registry (.memory/identities.yaml).

Anchored to the memory store so it is versioned alongside the tree. Contains public keys only, so committing it is safe. Callers may override the path.

Source code in memory/sharing/identity.py 
63
64
65
66
67
68
69
70
71
72
def default_registry_path() -> Path:
    """Default path of the shared identity registry (``.memory/identities.yaml``).

    Anchored to the memory store so it is versioned alongside the tree. Contains
    public keys only, so committing it is safe. Callers may override the path.
    """
    # Imported lazily to avoid a hard import-time dependency on storage globals.
    from memory import storage

    return storage.MEMORY_DIR / "identities.yaml"

load_local_identity 

load_local_identity(path: Path | str | None = None) -> LocalIdentity | None

Load this machine's identity, or None if the file is absent/invalid.

The public key is derived from the private key when the file omits it, so an identity file carrying only id + private_key still loads.

Source code in memory/sharing/identity.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def load_local_identity(path: Path | str | None = None) -> LocalIdentity | None:
    """Load this machine's identity, or ``None`` if the file is absent/invalid.

    The public key is derived from the private key when the file omits it, so an
    identity file carrying only ``id`` + ``private_key`` still loads.
    """
    src = Path(path) if path is not None else default_identity_path()
    if not src.exists():
        return None
    try:
        data = yaml.safe_load(src.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("could not read identity file %s: %s", src, exc)
        return None
    if not isinstance(data, Mapping):
        return None
    identity_id = str(data.get("id") or "").strip()
    private_key = str(data.get("private_key") or "").strip()
    if not identity_id or not private_key:
        return None
    public_key = str(data.get("public_key") or "").strip()
    if not public_key:
        try:
            public_key = str(x25519.Identity.from_str(private_key).to_public())
        except Exception as exc:  # pragma: no cover - malformed key
            logger.warning("identity file %s has an unusable private key: %s", src, exc)
            return None
    display_name = data.get("display_name")
    github = data.get("github")
    signing_public_key = data.get("signing_public_key")
    signing_private_key = data.get("signing_private_key")
    return LocalIdentity(
        id=identity_id,
        private_key=private_key,
        public_key=public_key,
        display_name=str(display_name) if display_name else None,
        github=str(github) if github else None,
        signing_public_key=str(signing_public_key) if signing_public_key else None,
        signing_private_key=str(signing_private_key) if signing_private_key else None,
    )

load_registry 

load_registry(path: Path | str | None = None) -> IdentityRegistry

Load the shared identity registry, or an empty one if absent/invalid.

Expected YAML shape::

identities:
  alice:
    public_key: age1...
    display_name: Alice
    github: alice-gh
    signing_public_key: ...   # optional Ed25519 signing public key (base64)
  bob: age1...          # shorthand: id -> public key (no signing key)

Malformed rows are skipped rather than raising, so a partially-authored registry still loads the good rows.

Source code in memory/sharing/identity.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def load_registry(path: Path | str | None = None) -> IdentityRegistry:
    """Load the shared identity registry, or an empty one if absent/invalid.

    Expected YAML shape::

        identities:
          alice:
            public_key: age1...
            display_name: Alice
            github: alice-gh
            signing_public_key: ...   # optional Ed25519 signing public key (base64)
          bob: age1...          # shorthand: id -> public key (no signing key)

    Malformed rows are skipped rather than raising, so a partially-authored
    registry still loads the good rows.
    """
    src = Path(path) if path is not None else default_registry_path()
    if not src.exists():
        return IdentityRegistry({})
    try:
        data = yaml.safe_load(src.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("could not read identity registry %s: %s", src, exc)
        return IdentityRegistry({})
    raw = data.get("identities") if isinstance(data, Mapping) else None
    if not isinstance(raw, Mapping):
        return IdentityRegistry({})
    entries: dict[str, RegistryEntry] = {}
    for entry_id, info in raw.items():
        eid = str(entry_id).strip()
        if not eid:
            continue
        parsed = _parse_registry_entry(eid, info)
        if parsed is not None:
            entries[eid] = parsed
    return IdentityRegistry(entries)

save_registry 

save_registry(registry: IdentityRegistry, path: Path | str | None = None) -> Path

Write the identity registry (public keys only) to disk.

Source code in memory/sharing/identity.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def save_registry(registry: IdentityRegistry, path: Path | str | None = None) -> Path:
    """Write the identity registry (public keys only) to disk."""
    dest = Path(path) if path is not None else default_registry_path()
    identities: dict[str, Any] = {}
    for entry_id, entry in registry.entries.items():
        row: dict[str, Any] = {"public_key": entry.public_key}
        if entry.display_name:
            row["display_name"] = entry.display_name
        if entry.github:
            row["github"] = entry.github
        if entry.signing_public_key:
            row["signing_public_key"] = entry.signing_public_key
        identities[entry_id] = row
    atomic_write(
        dest,
        yaml.dump({"identities": identities}, default_flow_style=False, sort_keys=True),
    )
    return dest

build_bundle 

build_bundle(output_dir: Path | str, *, registry: IdentityRegistry, memory_dir: Path | str | None = None, store_root: Path | str | None = None, overlay: Mapping[str, Any] | None = None, producer_id: str | None = None, federation_cache_root: Path | str | None = None, max_policy_age_hours: float | None = None) -> BuildResult

Filter synapse edges by both-ends grants, encrypt per recipient, write a bundle.

Pipeline (filtering happens BEFORE any encryption):

  1. Load the local .synapse overlay edges (overlay arg, else :func:zettelkasten.synapse.overlay.load_overlay).
  2. Build the memory entry_recipients and ZK note_recipients maps for the local workspace (reusing the two stores' sharing config loaders, firm expanded to the registry members). For any FEDERATED edge, also load the referenced peer's decrypted recipient-index policy from the federation cache (federation_cache_root, default the workspace cache).
  3. For each edge compute :func:edge_recipients — the INTERSECTION of its two endpoints' recipient sets (the ZK end resolved locally for a same-producer edge, or from the peer policy for a federated one). OMIT the edge when it resolves to no recipient (either side missing/ungranted, a federated endpoint with no resolvable peer policy, or an empty intersection) or when none of its recipients is registered. Never encrypt-then-drop.
  4. Encrypt each retained edge's placement-header envelope to its recipients' public keys and write units/<digest>.age.
  5. Write the cleartext manifest.json of {kind, path} pointers only.

This function OWNS output_dir/units and output_dir/manifest.json — the units directory is rebuilt from scratch every call so stale/revoked units never linger. Other files (e.g. .git) are left untouched. With no edges or no sharing policy it writes an empty bundle.

max_policy_age_hours is the OPT-IN fail-closed staleness TTL (default None = disabled). When set to a positive number and any cross-producer edge is present, this raises :class:StalePeerPolicyError — BEFORE any output is written — if a referenced peer's recipient policy is missing or older than the TTL, refusing to publish a possibly over-permissive stale edge set. Left unset, the default is unchanged: publish still ships (forward-only) and only emits a non-fatal staleness warning.

Source code in zettelkasten/synapse/sharing/bundle.py 
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
def build_bundle(
    output_dir: Path | str,
    *,
    registry: IdentityRegistry,
    memory_dir: Path | str | None = None,
    store_root: Path | str | None = None,
    overlay: Mapping[str, Any] | None = None,
    producer_id: str | None = None,
    federation_cache_root: Path | str | None = None,
    max_policy_age_hours: float | None = None,
) -> BuildResult:
    """Filter synapse edges by both-ends grants, encrypt per recipient, write a bundle.

    Pipeline (filtering happens BEFORE any encryption):

    1. Load the local ``.synapse`` overlay edges (``overlay`` arg, else
       :func:`zettelkasten.synapse.overlay.load_overlay`).
    2. Build the memory ``entry_recipients`` and ZK ``note_recipients`` maps for
       the local workspace (reusing the two stores' sharing config loaders,
       ``firm`` expanded to the registry members). For any FEDERATED edge, also
       load the referenced peer's decrypted recipient-index policy from the
       federation cache (``federation_cache_root``, default the workspace cache).
    3. For each edge compute :func:`edge_recipients` — the INTERSECTION of its
       two endpoints' recipient sets (the ZK end resolved locally for a
       same-producer edge, or from the peer policy for a federated one). OMIT the
       edge when it resolves to no recipient (either side missing/ungranted, a
       federated endpoint with no resolvable peer policy, or an empty
       intersection) or when none of its recipients is registered. Never
       encrypt-then-drop.
    4. Encrypt each retained edge's placement-header envelope to its recipients'
       public keys and write ``units/<digest>.age``.
    5. Write the cleartext ``manifest.json`` of ``{kind, path}`` pointers only.

    This function OWNS ``output_dir/units`` and ``output_dir/manifest.json`` — the
    units directory is rebuilt from scratch every call so stale/revoked units
    never linger. Other files (e.g. ``.git``) are left untouched. With no edges
    or no sharing policy it writes an empty bundle.

    ``max_policy_age_hours`` is the OPT-IN fail-closed staleness TTL (default
    ``None`` = disabled). When set to a positive number and any cross-producer edge
    is present, this raises :class:`StalePeerPolicyError` — BEFORE any output is
    written — if a referenced peer's recipient policy is missing or older than the
    TTL, refusing to publish a possibly over-permissive stale edge set. Left unset,
    the default is unchanged: publish still ships (forward-only) and only emits a
    non-fatal staleness warning.
    """
    from memory import storage

    output_dir = Path(output_dir)
    memory_dir = Path(memory_dir) if memory_dir is not None else storage.MEMORY_DIR
    if store_root is not None:
        store_root = Path(store_root)
    else:
        from zettelkasten.graph import GRAPHS_DIR
        store_root = GRAPHS_DIR

    if overlay is None:
        from zettelkasten.synapse import overlay as overlay_mod
        overlay = overlay_mod.load_overlay()
    edges = overlay.get("edges", []) if isinstance(overlay, Mapping) else []

    entry_recip = _memory_entry_recipients(memory_dir, registry)
    note_recip = _zk_note_recipients_map(store_root, registry)

    # Cross-producer edges resolve their ZK-end recipients from peer policies in
    # the federation cache (fail-closed if absent). Only touch the cache when a
    # federated edge is actually present, so the same-producer path is unchanged.
    has_federated = any(
        isinstance(e, Mapping) and ":" in str(e.get("zk_source") or "")
        for e in edges
    )
    peer_note_recip: dict[str, set[str]] = {}
    if has_federated:
        if federation_cache_root is not None:
            cache_root = Path(federation_cache_root)
        else:
            from zettelkasten.federation import (
                federation_cache_root as _fed_cache_root,
            )
            cache_root = _fed_cache_root()
        peer_note_recip = _peer_note_recipients(edges, cache_root)
        if max_policy_age_hours is not None and max_policy_age_hours > 0:
            # Opt-in HARD gate: refuse to publish on a stale/absent peer policy.
            # Raised BEFORE units/ is rebuilt (below), so an existing bundle is
            # left intact on failure. Deliberately NOT swallowed — aborting is the
            # whole point of the fail-closed TTL.
            _enforce_policy_ttl(
                edges, cache_root, timedelta(hours=float(max_policy_age_hours))
            )
        else:
            # Default: observability nudge (never fatal, never drops an edge). The
            # helper is purely diagnostic, so guard the call site: any unexpected
            # error inside it must NEVER propagate out of build_bundle / abort a
            # publish. Publish correctness must not depend on this observability aid.
            try:
                _warn_stale_peer_policies(edges, cache_root)
            except Exception as exc:  # pragma: no cover - defensive
                print(
                    "[synapse publish] warning: stale-peer-policy diagnostic failed "
                    f"({exc}); continuing publish (this is observability only).",
                    file=sys.stderr,
                )

    # ---- Prepare output (rebuild units/ from scratch) ---------------------
    units_dir = output_dir / UNITS_DIRNAME
    if units_dir.exists():
        shutil.rmtree(units_dir)
    units_dir.mkdir(parents=True, exist_ok=True)

    skipped_unregistered: set[str] = set()

    def _pubkeys_for(recips: set[str]) -> list[str]:
        keys: list[str] = []
        for rid in sorted(recips):
            pk = registry.public_key_for(rid)
            if pk:
                keys.append(pk)
            else:
                skipped_unregistered.add(rid)
        return keys

    units: list[BundleUnit] = []
    all_recipient_ids: set[str] = set()
    edge_count = 0

    for edge in edges:
        if not isinstance(edge, Mapping):
            continue
        recips = edge_recipients(edge, entry_recip, note_recip, peer_note_recip)
        if not recips:
            continue  # no both-ends recipient / unresolvable peer policy — omit
        pubkeys = _pubkeys_for(set(recips))
        if not pubkeys:
            logger.warning(
                "synapse edge %s:%s:%s is shared but no recipient is registered; omitting",
                edge.get("memory_id"), edge.get("zk_source"), edge.get("zk_id"),
            )
            continue
        memory_id = str(edge.get("memory_id") or "").strip()
        zk_source = str(edge.get("zk_source") or "").strip()
        zk_id = str(edge.get("zk_id") or "").strip()
        unit_key = f"{memory_id}:{zk_source}:{zk_id}"
        header = {
            "kind": KIND_SYNAPSE_EDGE,
            "memory_id": memory_id,
            "zk_source": zk_source,
            "zk_id": zk_id,
        }
        payload = json.dumps(
            _shareable_edge_payload(edge), sort_keys=True
        ).encode("utf-8")
        ciphertext = crypto.encrypt(_encode_envelope(header, payload), pubkeys)
        filename = _unit_filename(unit_key)
        atomic_write(units_dir / filename, ciphertext)
        units.append(BundleUnit(kind=KIND_SYNAPSE_EDGE, path=f"{UNITS_DIRNAME}/{filename}"))
        edge_count += 1
        all_recipient_ids.update(
            rid for rid in recips if registry.public_key_for(rid)
        )

    manifest = BundleManifest(
        format=BUNDLE_FORMAT,
        format_version=BUNDLE_FORMAT_VERSION,
        created_at=datetime.now(timezone.utc).isoformat(),
        units=tuple(units),
        producer_id=producer_id,
    )
    atomic_write(output_dir / MANIFEST_NAME, json.dumps(manifest.to_dict(), indent=2))

    return BuildResult(
        manifest=manifest,
        output_dir=output_dir,
        edge_unit_count=edge_count,
        recipient_ids=tuple(sorted(all_recipient_ids)),
        skipped_unregistered_ids=tuple(sorted(skipped_unregistered)),
    )

decrypt_bundle 

decrypt_bundle(bundle_dir: Path | str, out_root: Path | str, *, identity: LocalIdentity | Any | None = None, private_key: str | None = None) -> DecryptResult

Decrypt every edge this key can open and REBUILD a .synapse overlay cache.

Edges are decrypted and their records collected, then a valid overlay file is written to <out_root>/.synapse/links/links.json — a federation cache that is ENTIRELY bundle-owned, so a full rebuild-from-decrypted is correct: a revoked edge (dropped from a newer bundle, or no longer openable by this key) simply does not reappear. The rebuild is atomic.

Durability: a MISSING/CORRUPT manifest is treated as an empty bundle (the overlay is rebuilt to zero edges — full revocation). An UNREADABLE manifest, or ANY transient I/O fault reading a unit, ABORTS the rebuild WITHOUT touching the existing cache — a transient blip must never corrupt or erase a good overlay; the next clean sync rebuilds normally.

Provide the key via identity (a :class:LocalIdentity / pyrage identity), private_key (an AGE-SECRET-KEY-... string), or neither (to load ~/.angelo/identity.yaml).

Source code in zettelkasten/synapse/sharing/bundle.py 
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
def decrypt_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Decrypt every edge this key can open and REBUILD a ``.synapse`` overlay cache.

    Edges are decrypted and their records collected, then a valid overlay file is
    written to ``<out_root>/.synapse/links/links.json`` — a federation cache that
    is ENTIRELY bundle-owned, so a full rebuild-from-decrypted is correct: a
    revoked edge (dropped from a newer bundle, or no longer openable by this key)
    simply does not reappear. The rebuild is atomic.

    Durability: a MISSING/CORRUPT manifest is treated as an empty bundle (the
    overlay is rebuilt to zero edges — full revocation). An UNREADABLE manifest,
    or ANY transient I/O fault reading a unit, ABORTS the rebuild WITHOUT touching
    the existing cache — a transient blip must never corrupt or erase a good
    overlay; the next clean sync rebuilds normally.

    Provide the key via ``identity`` (a :class:`LocalIdentity` / pyrage identity),
    ``private_key`` (an ``AGE-SECRET-KEY-...`` string), or neither (to load
    ``~/.angelo/identity.yaml``).
    """
    bundle_dir = Path(bundle_dir)
    out_root = Path(out_root)
    overlay_path = _materialized_overlay_path(out_root)
    key = _resolve_private_key(identity, private_key)

    try:
        manifest = read_manifest(bundle_dir)
    except FileNotFoundError as exc:
        logger.warning("bundle manifest absent (%s); rebuilding overlay as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except (json.JSONDecodeError, ValueError) as exc:
        logger.warning("bundle manifest corrupt (%s); rebuilding overlay as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except OSError as exc:
        logger.warning(
            "bundle manifest unreadable due to transient I/O (%s); aborting decrypt "
            "without touching the cache to preserve the valid overlay", exc
        )
        return DecryptResult(output_dir=overlay_path.parent, edge_count=0, total_units=0)

    edges: list[dict[str, Any]] = []
    transient_io_error = False

    for unit in manifest.units:
        unit_path = _resolve_unit_path(bundle_dir, unit)
        if unit_path is None:
            continue
        if not unit_path.exists():
            # Genuinely gone from disk (e.g. revoked): a full rebuild correctly
            # omits it. Not a transient fault.
            logger.warning("bundle unit missing on disk: %s", unit_path)
            continue
        try:
            ciphertext = unit_path.read_bytes()
        except OSError as exc:
            # Present but unreadable — a transient I/O fault. Rebuilding now could
            # drop a still-valid edge and clobber a good overlay, so abort the
            # rebuild entirely this run (preserve the cache) and retry next sync.
            logger.warning(
                "could not read bundle unit %s (transient I/O): %s; "
                "aborting overlay rebuild this run to preserve the cache",
                unit_path, exc,
            )
            transient_io_error = True
            break
        try:
            plaintext = crypto.decrypt(ciphertext, key)
        except crypto.DecryptError:
            continue  # not encrypted to this key
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("failed to decrypt unit %s: %s", unit.path, exc)
            continue

        decoded = _decode_envelope(plaintext)
        if decoded is None:
            logger.warning("decrypted unit %s has a malformed envelope; skipping", unit.path)
            continue
        header, payload = decoded
        placement = _valid_placement(header)
        if placement is None:
            logger.warning("decrypted unit %s has an invalid placement; skipping", unit.path)
            continue
        memory_id, zk_source, zk_id = placement
        try:
            edge = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            logger.warning("decrypted unit %s has a malformed edge payload; skipping", unit.path)
            continue
        if not isinstance(edge, dict):
            continue
        # Trust the validated placement header for the id-bearing fields so a
        # rebuilt edge always carries consistent, contained ids.
        edge["memory_id"] = memory_id
        edge["zk_source"] = zk_source
        edge["zk_id"] = zk_id
        edges.append(edge)

    if transient_io_error:
        return DecryptResult(output_dir=overlay_path.parent, edge_count=0, total_units=len(manifest.units))

    overlay = {
        "version": OVERLAY_FORMAT_VERSION,
        "manifest": {
            "source": BUNDLE_FORMAT,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "edge_count": len(edges),
        },
        "edges": edges,
    }
    atomic_write(overlay_path, json.dumps(overlay, indent=2, ensure_ascii=False).encode("utf-8"))

    return DecryptResult(
        output_dir=overlay_path.parent,
        edge_count=len(edges),
        total_units=len(manifest.units),
    )

open_bundle 

open_bundle(bundle_dir: Path | str, out_root: Path | str, *, identity: LocalIdentity | Any | None = None, private_key: str | None = None) -> DecryptResult

Alias for :func:decrypt_bundle.

Source code in zettelkasten/synapse/sharing/bundle.py 
1113
1114
1115
1116
1117
1118
1119
1120
1121
def open_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Alias for :func:`decrypt_bundle`."""
    return decrypt_bundle(bundle_dir, out_root, identity=identity, private_key=private_key)

read_manifest 

read_manifest(bundle_dir: Path | str) -> BundleManifest

Read and parse a bundle's cleartext manifest.json.

Source code in zettelkasten/synapse/sharing/bundle.py 
421
422
423
424
425
def read_manifest(bundle_dir: Path | str) -> BundleManifest:
    """Read and parse a bundle's cleartext ``manifest.json``."""
    path = Path(bundle_dir) / MANIFEST_NAME
    data = json.loads(path.read_text(encoding="utf-8"))
    return BundleManifest.from_dict(data)

edge_recipients 

edge_recipients(edge: Mapping[str, Any], entry_recipients_map: Mapping[str, set[str]], note_recipients_map: Mapping[str, set[str]], peer_note_recipients_map: Mapping[str, set[str]] | None = None) -> set[str]

Recipients who may see a synapse edge: the INTERSECTION of both endpoints.

An edge names a memory entry (memory_id) and a ZK note (zk_source/zk_id). It reaches a recipient R iff R can open BOTH — so the recipient set is entry_recipients_map[memory_id] ∩ (that ZK note's recipients). This intersection is the whole id-secrecy property: an edge names ids from two stores, so shipping it to anyone who lacks EITHER end would leak an id they were never granted.

The ZK note's recipients come from one of two maps, keyed by "<zk_source>/<zk_id>":

  • SAME-PRODUCER edge (zk_source has no ":") — the local note_recipients_map, exactly as before. peer_note_recipients_map is ignored, so same-producer behaviour is UNCHANGED.
  • CROSS-PRODUCER / FEDERATED edge (zk_source contains ":", e.g. "remote:alpha") — the ZK end lives in a peer repo, so its recipients are looked up in the supplied peer_note_recipients_map (a decrypted recipient-index export; see :mod:zettelkasten.synapse.sharing.recipient_index).

Returns the EMPTY set (→ the caller omits the edge) when:

  • memory_id, zk_id or zk_source is missing/blank;
  • the edge is federated and peer_note_recipients_map is absent/empty or does not contain the note — FAIL CLOSED (the default when no peer policy is supplied, preserving today's "omit federated" behaviour);
  • either endpoint is unresolvable (absent from its recipient map, i.e. private/ungranted); or
  • the intersection itself is empty (no recipient sees both ends).
Source code in zettelkasten/synapse/sharing/config.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def edge_recipients(
    edge: Mapping[str, Any],
    entry_recipients_map: Mapping[str, set[str]],
    note_recipients_map: Mapping[str, set[str]],
    peer_note_recipients_map: Mapping[str, set[str]] | None = None,
) -> set[str]:
    """Recipients who may see a synapse edge: the INTERSECTION of both endpoints.

    An edge names a memory entry (``memory_id``) and a ZK note
    (``zk_source``/``zk_id``). It reaches a recipient R iff R can open BOTH — so
    the recipient set is ``entry_recipients_map[memory_id]`` ∩ (that ZK note's
    recipients). This intersection is the whole id-secrecy property: an edge names
    ids from two stores, so shipping it to anyone who lacks EITHER end would leak
    an id they were never granted.

    The ZK note's recipients come from one of two maps, keyed by
    ``"<zk_source>/<zk_id>"``:

    * SAME-PRODUCER edge (``zk_source`` has no ``":"``) — the local
      ``note_recipients_map``, exactly as before. ``peer_note_recipients_map`` is
      ignored, so same-producer behaviour is UNCHANGED.
    * CROSS-PRODUCER / FEDERATED edge (``zk_source`` contains ``":"``, e.g.
      ``"remote:alpha"``) — the ZK end lives in a peer repo, so its recipients are
      looked up in the supplied ``peer_note_recipients_map`` (a decrypted
      recipient-index export; see
      :mod:`zettelkasten.synapse.sharing.recipient_index`).

    Returns the EMPTY set (→ the caller omits the edge) when:

    * ``memory_id``, ``zk_id`` or ``zk_source`` is missing/blank;
    * the edge is federated and ``peer_note_recipients_map`` is absent/empty or
      does not contain the note — FAIL CLOSED (the default when no peer policy is
      supplied, preserving today's "omit federated" behaviour);
    * either endpoint is unresolvable (absent from its recipient map,
      i.e. private/ungranted); or
    * the intersection itself is empty (no recipient sees both ends).
    """
    if not isinstance(edge, Mapping):
        return set()
    memory_id = str(edge.get("memory_id") or "").strip()
    zk_id = str(edge.get("zk_id") or "").strip()
    zk_source = str(edge.get("zk_source") or "").strip()
    if not (memory_id and zk_id and zk_source):
        return set()
    note_key = f"{zk_source}/{zk_id}"
    if ":" in zk_source:
        # A federated endpoint (``<repo_id>:<graph>``) is not locally resolvable:
        # its recipients come from the peer-policy map, and absent that map (or
        # the note within it) the edge is omitted — fail-closed.
        if not peer_note_recipients_map:
            return set()
        note_recips = peer_note_recipients_map.get(note_key)
    else:
        note_recips = note_recipients_map.get(note_key)
    mem_recips = entry_recipients_map.get(memory_id)
    if not mem_recips or not note_recips:
        return set()
    return set(mem_recips) & set(note_recips)

parse_sharing_config 

parse_sharing_config(config: Mapping[str, Any] | None) -> SynapseSharingConfig

Parse the sharing block + bundle_subscriptions of a config dict.

config is the plain dict read from .synapse/config.yaml (via :func:read_config). Missing/malformed blocks yield an empty config (nothing to publish or subscribe) — never raises on a partial config. The existing intersects: block is ignored here (owned by :mod:zettelkasten.synapse.config).

Source code in zettelkasten/synapse/sharing/config.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def parse_sharing_config(config: Mapping[str, Any] | None) -> SynapseSharingConfig:
    """Parse the ``sharing`` block + ``bundle_subscriptions`` of a config dict.

    ``config`` is the plain dict read from ``.synapse/config.yaml`` (via
    :func:`read_config`). Missing/malformed blocks yield an empty config
    (nothing to publish or subscribe) — never raises on a partial config. The
    existing ``intersects:`` block is ignored here (owned by
    :mod:`zettelkasten.synapse.config`).
    """
    if not isinstance(config, Mapping):
        return SynapseSharingConfig()

    block = config.get(SHARING_KEY)
    block = block if isinstance(block, Mapping) else {}
    bundle_repo = str(block.get("bundle_repo") or "").strip() or None
    producer_id = str(block.get("producer_id") or "").strip() or None

    # Opt-in fail-closed TTL: only a POSITIVE finite number enables it; anything
    # else (missing, non-numeric, bool, <= 0) leaves it disabled (None) so a
    # malformed value silently degrades to today's warn-only behaviour rather than
    # accidentally blocking every publish.
    max_age = None
    raw_ttl = block.get("recipient_index_max_age_hours")
    if raw_ttl is not None and not isinstance(raw_ttl, bool):
        try:
            val = float(raw_ttl)
        except (TypeError, ValueError):
            val = None
        if val is not None and val > 0:
            max_age = val

    raw_subs = config.get(BUNDLE_SUBSCRIPTIONS_KEY)
    subs: tuple[Mapping[str, Any], ...] = ()
    if isinstance(raw_subs, list):
        subs = tuple(s for s in raw_subs if isinstance(s, Mapping))

    return SynapseSharingConfig(
        bundle_repo=bundle_repo,
        producer_id=producer_id,
        bundle_subscriptions=subs,
        recipient_index_max_age_hours=max_age,
    )

read_config 

read_config() -> dict[str, Any]

Read the synapse config, returning {} when missing or malformed.

Source code in zettelkasten/synapse/config.py 
146
147
148
149
150
151
152
153
154
155
156
def read_config() -> dict[str, Any]:
    """Read the synapse config, returning ``{}`` when missing or malformed."""
    path = config_path()
    if not path.exists():
        return {}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("Could not read synapse config %s: %s", path, exc)
        return {}
    return data if isinstance(data, dict) else {}

build_recipient_index 

build_recipient_index(output_dir: Path | str, note_recipients: Mapping[str, set[str]], registry: IdentityRegistry, *, producer_id: str | None = None, signing_private_key: str | None = None) -> RecipientIndexBuildResult

Write the firm-encrypted recipient-index export as a ZK-bundle sibling.

note_recipients maps a "<box>/<note_id>" key to the recipient id set for EXACTLY the notes shipped in the ZK bundle (pass the already-resolved per-note recipients — do NOT invent a new scope). The mapping is encrypted once to ALL registered firm members (registry.member_ids()), so any firm member can decrypt it (the accepted, firm-bounded membership leak).

Owns output_dir/recipient-index.json and output_dir/recipient-index/: both are removed first and rewritten from scratch, so a revoked note never lingers. With an empty scope (no shared notes) or no registered members, the artifacts are simply left removed and written=False is returned.

Signing (tamper-evident, non-repudiable binding to the producer): when signing_private_key is given, the EXACT canonical plaintext bytes that get encrypted (json.dumps(mapping, sort_keys=True).encode()) are Ed25519-signed and the base64 signature is attached to the cleartext manifest alongside producer_id (the verifier resolves the producer's signing pubkey by producer_id). With no signing key the export is built UNSIGNED — a warning is logged, but the publish never fails (backward-compat).

Source code in zettelkasten/synapse/sharing/recipient_index.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
def build_recipient_index(
    output_dir: Path | str,
    note_recipients: Mapping[str, set[str]],
    registry: IdentityRegistry,
    *,
    producer_id: str | None = None,
    signing_private_key: str | None = None,
) -> RecipientIndexBuildResult:
    """Write the firm-encrypted recipient-index export as a ZK-bundle sibling.

    ``note_recipients`` maps a ``"<box>/<note_id>"`` key to the recipient id set
    for EXACTLY the notes shipped in the ZK bundle (pass the already-resolved
    per-note recipients — do NOT invent a new scope). The mapping is encrypted
    once to ALL registered firm members (``registry.member_ids()``), so any firm
    member can decrypt it (the accepted, firm-bounded membership leak).

    Owns ``output_dir/recipient-index.json`` and ``output_dir/recipient-index/``:
    both are removed first and rewritten from scratch, so a revoked note never
    lingers. With an empty scope (no shared notes) or no registered members, the
    artifacts are simply left removed and ``written=False`` is returned.

    Signing (tamper-evident, non-repudiable binding to the producer):
    when ``signing_private_key`` is given, the EXACT canonical ``plaintext`` bytes
    that get encrypted (``json.dumps(mapping, sort_keys=True).encode()``) are
    Ed25519-signed and the base64 ``signature`` is attached to the cleartext
    manifest alongside ``producer_id`` (the verifier resolves the producer's
    signing pubkey by ``producer_id``). With no signing key the export is built
    UNSIGNED — a warning is logged, but the publish never fails (backward-compat).
    """
    output_dir = Path(output_dir)
    units_dir = output_dir / UNITS_DIRNAME
    manifest_path = output_dir / MANIFEST_NAME

    # Rebuild-from-scratch: drop any prior export so revoked entries never linger.
    if units_dir.exists():
        shutil.rmtree(units_dir)
    _unlink_quietly(manifest_path)

    mapping: dict[str, list[str]] = {}
    for key, recips in note_recipients.items():
        k = str(key).strip()
        if not k or "/" not in k:
            continue
        ids = sorted({str(r).strip() for r in recips if str(r).strip()})
        if ids:
            mapping[k] = ids

    if not mapping:
        return RecipientIndexBuildResult(
            output_dir=output_dir, note_count=0, recipient_ids=(),
            skipped_unregistered_ids=(), written=False,
        )

    member_ids = sorted(registry.member_ids())
    pubkeys: list[str] = []
    skipped: set[str] = set()
    for rid in member_ids:
        pk = registry.public_key_for(rid)
        if pk:
            pubkeys.append(pk)
        else:
            skipped.add(rid)
    if not pubkeys:
        logger.warning(
            "recipient-index export: no registered firm members to encrypt to; "
            "omitting export"
        )
        return RecipientIndexBuildResult(
            output_dir=output_dir, note_count=0, recipient_ids=(),
            skipped_unregistered_ids=tuple(sorted(skipped)), written=False,
        )

    # The signed bytes MUST be EXACTLY the canonical mapping bytes that get
    # encrypted below — the verifier signs/verifies the same ``plaintext``.
    plaintext = json.dumps(mapping, sort_keys=True).encode("utf-8")

    signature: str | None = None
    if signing_private_key:
        try:
            signature = signing.sign(plaintext, signing_private_key)
        except ValueError as exc:
            # A malformed/empty signing key -> publish UNSIGNED rather than fail.
            logger.warning(
                "recipient-index export: signing key unusable (%s); publishing "
                "UNSIGNED", exc
            )
            signature = None
    else:
        logger.warning(
            "recipient-index export: no signing key available; publishing UNSIGNED "
            "(peers with a registered signing key for this producer will REJECT it)"
        )

    ciphertext = crypto.encrypt(plaintext, pubkeys)
    units_dir.mkdir(parents=True, exist_ok=True)
    filename = _unit_filename()
    atomic_write(units_dir / filename, ciphertext)

    manifest: dict[str, Any] = {
        "format": EXPORT_FORMAT,
        "format_version": EXPORT_FORMAT_VERSION,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "unit": f"{UNITS_DIRNAME}/{filename}",
    }
    if producer_id:
        manifest["producer_id"] = str(producer_id)
    if signature:
        manifest["signature"] = signature
    atomic_write(manifest_path, json.dumps(manifest, indent=2))

    return RecipientIndexBuildResult(
        output_dir=output_dir,
        note_count=len(mapping),
        recipient_ids=tuple(rid for rid in member_ids if rid not in skipped),
        skipped_unregistered_ids=tuple(sorted(skipped)),
        written=True,
        signed=signature is not None,
    )

decrypt_recipient_index 

decrypt_recipient_index(bundle_dir: Path | str, out_root: Path | str, *, identity: LocalIdentity | Any | None = None, private_key: str | None = None, registry: IdentityRegistry | None = None, expected_producer_id: str | None = None) -> RecipientIndexDecryptResult

Decrypt a peer's recipient-index export into its federation cache.

Reads <bundle_dir>/recipient-index.json + its ciphertext, decrypts with the local firm key, defensively validates the (untrusted) mapping, and writes <out_root>/.zettelkasten/_policy/note-recipients.json.

Signature verification (fail-closed, opt-in via registry): when registry is provided the Ed25519 signature over the decrypted canonical plaintext bytes is enforced against the producer's registered signing public key. expected_producer_id (the subscription/peer id the consumer believes it is syncing) binds the manifest to that peer:

  • expected_producer_id given (the CLI federation sync path): the manifest's producer_id MUST equal expected_producer_id (a missing or relabelled producer_id is rejected as TAMPER), and the signing key is resolved by expected_producer_id — never the manifest's self-asserted value. Then: registered key + signature -> must verify (else TAMPER); registered key + NO signature -> :class:RecipientIndexSignatureError (DOWNGRADE); NO registered key -> accept UNSIGNED (weaker mode, logged).
  • expected_producer_id None (API back-compat): the signing key is resolved by the manifest's own producer_id (a no-op when the manifest carries none), preserving the pre-P1 behaviour for existing registry= callers.
  • registry is None -> verification is SKIPPED entirely (pure backward compat; existing callers are unaffected).

A verification failure (TAMPER/DOWNGRADE/relabel) is a HARD failure: the :class:RecipientIndexSignatureError propagates AND any stale policy for this peer is REMOVED first, so a rejected index never leaves a stale/partial policy file behind (the edge then fails closed).

signing.verify() itself never raises (returns False); this function raises on a False result so a tampered/downgraded export fails closed.

Fail-safe / forward-only revocation:

  • MISSING manifest (peer no longer exports) or a mapping this key cannot open / that validates empty -> the stale policy file (and its stamp) is REMOVED (revocation).
  • a corrupt/unsafe manifest, a bad unit path, or a missing unit -> stale policy removed (nothing valid to trust).
  • a TRANSIENT I/O fault reading the manifest/unit -> the existing policy is PRESERVED untouched and retried next sync.

On success the policy is (over)written AND a freshness stamp sidecar (:data:POLICY_STAMP_FILENAME) is written carrying the producer's manifest created_at/producer_id and this decrypt's timestamp, so a downstream caller can observe how old the policy it trusts is.

Source code in zettelkasten/synapse/sharing/recipient_index.py 
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
def decrypt_recipient_index(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
    registry: IdentityRegistry | None = None,
    expected_producer_id: str | None = None,
) -> RecipientIndexDecryptResult:
    """Decrypt a peer's recipient-index export into its federation cache.

    Reads ``<bundle_dir>/recipient-index.json`` + its ciphertext, decrypts with
    the local firm key, defensively validates the (untrusted) mapping, and writes
    ``<out_root>/.zettelkasten/_policy/note-recipients.json``.

    Signature verification (fail-closed, opt-in via ``registry``):
    when ``registry`` is provided the Ed25519 signature over the decrypted
    canonical ``plaintext`` bytes is enforced against the producer's registered
    signing public key. ``expected_producer_id`` (the subscription/peer id the
    consumer believes it is syncing) binds the manifest to that peer:

    * ``expected_producer_id`` given (the CLI ``federation sync`` path): the
      manifest's ``producer_id`` MUST equal ``expected_producer_id`` (a missing or
      relabelled producer_id is rejected as TAMPER), and the signing key is
      resolved by ``expected_producer_id`` — never the manifest's self-asserted
      value. Then: registered key + signature -> must verify (else TAMPER);
      registered key + NO signature -> :class:`RecipientIndexSignatureError`
      (DOWNGRADE); NO registered key -> accept UNSIGNED (weaker mode, logged).
    * ``expected_producer_id`` None (API back-compat): the signing key is resolved
      by the manifest's own ``producer_id`` (a no-op when the manifest carries
      none), preserving the pre-P1 behaviour for existing ``registry=`` callers.
    * ``registry is None`` -> verification is SKIPPED entirely (pure backward
      compat; existing callers are unaffected).

    A verification failure (TAMPER/DOWNGRADE/relabel) is a HARD failure: the
    :class:`RecipientIndexSignatureError` propagates AND any stale policy for this
    peer is REMOVED first, so a rejected index never leaves a stale/partial policy
    file behind (the edge then fails closed).

    ``signing.verify()`` itself never raises (returns ``False``); this function
    raises on a ``False`` result so a tampered/downgraded export fails closed.

    Fail-safe / forward-only revocation:

    * MISSING manifest (peer no longer exports) or a mapping this key cannot open
      / that validates empty -> the stale policy file (and its stamp) is REMOVED
      (revocation).
    * a corrupt/unsafe manifest, a bad unit path, or a missing unit -> stale
      policy removed (nothing valid to trust).
    * a TRANSIENT I/O fault reading the manifest/unit -> the existing policy is
      PRESERVED untouched and retried next sync.

    On success the policy is (over)written AND a freshness stamp sidecar
    (:data:`POLICY_STAMP_FILENAME`) is written carrying the producer's manifest
    ``created_at``/``producer_id`` and this decrypt's timestamp, so a downstream
    caller can observe how old the policy it trusts is.
    """
    bundle_dir = Path(bundle_dir)
    policy_path = _policy_path(out_root)
    manifest_path = bundle_dir / MANIFEST_NAME

    try:
        raw = manifest_path.read_text(encoding="utf-8")
    except FileNotFoundError:
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)
    except OSError as exc:
        logger.warning(
            "recipient-index manifest unreadable (transient I/O): %s; preserving "
            "existing policy", exc
        )
        return RecipientIndexDecryptResult(policy_path.parent, 0)

    try:
        manifest = json.loads(raw)
    except (json.JSONDecodeError, ValueError):
        logger.warning("recipient-index manifest corrupt; removing stale policy")
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)
    if not isinstance(manifest, dict):
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)

    unit_path = _resolve_unit_path(bundle_dir, str(manifest.get("unit") or ""))
    if unit_path is None:
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)
    if not unit_path.exists():
        logger.warning("recipient-index unit missing on disk: %s", unit_path)
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)
    try:
        ciphertext = unit_path.read_bytes()
    except OSError as exc:
        logger.warning(
            "recipient-index unit unreadable (transient I/O): %s; preserving "
            "existing policy", exc
        )
        return RecipientIndexDecryptResult(policy_path.parent, 0)

    key = _resolve_private_key(identity, private_key)
    try:
        plaintext = crypto.decrypt(ciphertext, key)
    except crypto.DecryptError:
        # Not a firm member (cannot open the firm-wide export): no valid policy.
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("recipient-index decrypt failed: %s", exc)
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)

    # Fail-closed signature verification (opt-in via ``registry``). Enforced over
    # the EXACT decrypted ``plaintext`` bytes — the same canonical mapping bytes
    # the producer signed. This raises (fail-closed) BEFORE the mapping is trusted
    # / persisted, so a tampered or downgraded export never reaches
    # _validate_mapping. On rejection the stale policy is REMOVED before the error
    # propagates, so a rejected index leaves NO stale/partial policy behind.
    try:
        _verify_signature(
            manifest, plaintext, registry,
            expected_producer_id=expected_producer_id,
        )
    except RecipientIndexSignatureError:
        _remove_policy(policy_path)
        raise

    mapping = _validate_mapping(plaintext)
    if not mapping:
        _remove_policy(policy_path)
        return RecipientIndexDecryptResult(policy_path.parent, 0)

    atomic_write(
        policy_path,
        json.dumps(mapping, sort_keys=True, indent=2).encode("utf-8"),
    )
    # Freshness stamp (best-effort): a plain sidecar so the policy file's shape is
    # untouched. Carries the producer's generation (manifest ``created_at`` /
    # ``producer_id``) plus this decrypt's timestamp and note count.
    stamp: dict[str, Any] = {
        "format": EXPORT_FORMAT,
        "format_version": EXPORT_FORMAT_VERSION,
        "decrypted_at": datetime.now(timezone.utc).isoformat(),
        "note_count": len(mapping),
    }
    created_at = manifest.get("created_at")
    if isinstance(created_at, str) and created_at.strip():
        stamp["created_at"] = created_at
    producer_id = manifest.get("producer_id")
    if isinstance(producer_id, str) and producer_id.strip():
        stamp["producer_id"] = producer_id
    try:
        atomic_write(
            _stamp_path(policy_path),
            json.dumps(stamp, sort_keys=True, indent=2).encode("utf-8"),
        )
    except OSError as exc:  # pragma: no cover - defensive
        logger.warning("could not write recipient-index freshness stamp: %s", exc)
    return RecipientIndexDecryptResult(policy_path.parent, len(mapping))

load_recipient_index 

load_recipient_index(cache_root: Path | str, peer_id: str) -> dict[str, set[str]] | None

Load a peer's DECRYPTED recipient-index policy from the federation cache.

Reads <cache_root>/<peer_id>/.zettelkasten/_policy/note-recipients.json, re-validates it defensively (belt-and-suspenders over the decrypt-time validation), and returns a {"<box>/<note_id>": {recipient_ids}} map — or None when absent/unreadable/empty/invalid (the caller then FAILS CLOSED, omitting the cross-producer edge).

peer_id is derived from an edge's zk_source and is ID_RE-validated before it is turned into a path (defence-in-depth over :func:safe_join); a malformed peer id fails closed with None.

Source code in zettelkasten/synapse/sharing/recipient_index.py 
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
def load_recipient_index(
    cache_root: Path | str, peer_id: str
) -> dict[str, set[str]] | None:
    """Load a peer's DECRYPTED recipient-index policy from the federation cache.

    Reads ``<cache_root>/<peer_id>/.zettelkasten/_policy/note-recipients.json``,
    re-validates it defensively (belt-and-suspenders over the decrypt-time
    validation), and returns a ``{"<box>/<note_id>": {recipient_ids}}`` map — or
    ``None`` when absent/unreadable/empty/invalid (the caller then FAILS CLOSED,
    omitting the cross-producer edge).

    ``peer_id`` is derived from an edge's ``zk_source`` and is ID_RE-validated
    before it is turned into a path (defence-in-depth over :func:`safe_join`); a
    malformed peer id fails closed with ``None``.
    """
    pid = _valid_peer_id(peer_id)
    if pid is None:
        return None
    try:
        policy_path = safe_join(
            Path(cache_root), pid, ".zettelkasten", POLICY_DIRNAME, POLICY_FILENAME
        )
    except ValueError:
        return None
    try:
        raw = policy_path.read_bytes()
    except (FileNotFoundError, OSError):
        return None
    mapping = _validate_mapping(raw)
    if not mapping:
        return None
    return {key: set(ids) for key, ids in mapping.items()}