registry

registry

The devices this installation knows, and what each is allowed to do.

The registry answers the three questions the domain cannot (spec §5): which devices exist, what each one is, and whether it may still be listened to. Succès remains the single source of truth for tasks and notes — nothing here duplicates business data.

Pairing mirrors the mechanism Succès already proved in production: a one-time invitation token, hashed at rest, short TTL, redeemed exactly once. What the mesh adds is the exchange of PUBLIC KEYS, which is what later lets a command be verified rather than merely accepted.

Trust is a state, not a flag (spec §5): UNTRUSTED → PENDING → TRUSTED → REVOKED, and revocation is terminal — a revoked device is never silently re-admitted by presenting the same key.

Classes

MeshError

Bases: RuntimeError

A mesh operation the user must see explained, in French.

DeviceRegistry

DeviceRegistry(db_path: str | Path | None = None)

Devices known to this installation, in its own SQLite file.

Source code in src/diapason/mesh/registry.py

def __init__(self, db_path: str | Path | None = None) -> None:
    self.db_path = Path(db_path or (get_data_dir() / "mesh.db"))
    self.db_path.parent.mkdir(parents=True, exist_ok=True)
    with closing(self._connect()) as conn, conn:
        conn.executescript(_SCHEMA)
        self._ensure_columns(conn)
        conn.commit()

Methods:

create_pairing

create_pairing(device_name: str) -> dict[str, Any]

Open a one-time, short-lived invitation for a new device.

Source code in src/diapason/mesh/registry.py

def create_pairing(self, device_name: str) -> dict[str, Any]:
    """Open a one-time, short-lived invitation for a new device."""
    name = _clean(device_name, field="Le nom de l'appareil", maximum=80)
    token = f"diapason_mesh_{secrets.token_urlsafe(32)}"
    created = now_ms()
    expires = created + PAIRING_TTL_MS
    with closing(self._connect()) as conn, conn:
        # Expired and spent invitations are swept on each new one: an
        # unbounded table of dead secrets is a liability, not a record.
        conn.execute(
            "DELETE FROM mesh_pairings "
            "WHERE expires_at_ms < ? OR redeemed_at_ms IS NOT NULL",
            (created,),
        )
        conn.execute(
            "INSERT INTO mesh_pairings"
            "(token_hash,device_name,created_at_ms,expires_at_ms)"
            " VALUES (?,?,?,?)",
            (_token_hash(token), name, created, expires),
        )
        conn.commit()
    return {
        "pairingToken": token,
        "deviceName": name,
        "expiresAtMs": expires,
        "expiresInSeconds": PAIRING_TTL_MS // 1000,
    }

redeem_pairing

redeem_pairing(
    token: str,
    *,
    device_id: str,
    public_key_b64: str,
    name: str,
    platform: str,
    device_type: str = "DESKTOP",
    declared_capabilities: Sequence[str] | None = None,
    app_version: str = "",
) -> dict[str, Any]

Spend an invitation and enrol the device that presented it.

The invitation proves a human authorised THIS enrolment; the public key is what every later command will be checked against.

Source code in src/diapason/mesh/registry.py

def redeem_pairing(
    self,
    token: str,
    *,
    device_id: str,
    public_key_b64: str,
    name: str,
    platform: str,
    device_type: str = "DESKTOP",
    declared_capabilities: Sequence[str] | None = None,
    app_version: str = "",
) -> dict[str, Any]:
    """Spend an invitation and enrol the device that presented it.

    The invitation proves a human authorised THIS enrolment; the public
    key is what every later command will be checked against.
    """
    if not token.startswith("diapason_mesh_"):
        raise MeshError("Ce code d'appairage est invalide.")
    device_id = _clean(device_id, field="L'identifiant d'appareil", maximum=120)
    name = _clean(name, field="Le nom de l'appareil", maximum=80)
    platform = _clean(platform, field="La plateforme", maximum=40).upper()
    device_type = (device_type or "DESKTOP").strip().upper()
    if device_type not in _DEVICE_TYPES:
        raise MeshError(f"Le type d'appareil « {device_type} » est inconnu.")
    key = self._validate_public_key(public_key_b64)

    stamp = now_ms()
    with closing(self._connect()) as conn, conn:
        # Claim the invitation FIRST, and let the UPDATE be the test.
        #
        # Reading `redeemed_at_ms` and then writing it is two steps with a
        # gap, and sqlite3 opens its transaction only at the first write —
        # so two redemptions racing both read "unused" and both succeed,
        # enrolling two devices from one invitation. Worse than the extra
        # device is the silence: the legitimate one still succeeds, so the
        # « déjà utilisé » message that would have warned the user their
        # code was stolen never appears.
        #
        # Making the spend the guard is the pattern this module already
        # uses for nonces and for beacon watermarks: only one writer can
        # see the old value, so only one can win.
        claimed = conn.execute(
            "UPDATE mesh_pairings SET redeemed_at_ms=? "
            "WHERE token_hash=? AND redeemed_at_ms IS NULL "
            "  AND expires_at_ms >= ?",
            (stamp, _token_hash(token), stamp),
        )
        if claimed.rowcount != 1:
            # Say which of the three it was, since the user acts on it
            # differently: mistyped, already used, or waited too long.
            row = conn.execute(
                "SELECT expires_at_ms, redeemed_at_ms FROM mesh_pairings "
                "WHERE token_hash=?",
                (_token_hash(token),),
            ).fetchone()
            if row is None:
                raise MeshError("Ce code d'appairage est inconnu.")
            if row["redeemed_at_ms"] is not None:
                raise MeshError("Ce code d'appairage a déjà été utilisé.")
            raise MeshError("Ce code d'appairage a expiré.")

        existing = conn.execute(
            "SELECT public_key, trust_level FROM mesh_devices WHERE device_id=?",
            (device_id,),
        ).fetchone()
        if existing is not None:
            # Revocation is terminal: presenting a fresh invitation must
            # not launder a device the user deliberately cut off.
            if existing["trust_level"] == TRUST_REVOKED:
                raise MeshError(
                    "Cet appareil a été révoqué. Supprimez-le d'abord de "
                    "la liste des appareils pour pouvoir le réappairer."
                )
            # A known device that turns up with a DIFFERENT key is either
            # a reinstall or an impostor; either way the human decides.
            if existing["public_key"] != key:
                raise MeshError(
                    "Un appareil portant cet identifiant est déjà connu "
                    "avec une autre clé. Révoquez-le avant de le réappairer."
                )

        conn.execute(
            """INSERT INTO mesh_devices
               (device_id, public_key, name, platform, device_type,
                trust_level, declared_capabilities, app_version,
                created_at_ms, last_seen_at_ms)
               VALUES (?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(device_id) DO UPDATE SET
                 name=excluded.name,
                 platform=excluded.platform,
                 device_type=excluded.device_type,
                 trust_level=excluded.trust_level,
                 declared_capabilities=excluded.declared_capabilities,
                 app_version=excluded.app_version,
                 last_seen_at_ms=excluded.last_seen_at_ms,
                 -- Un ré-appairage repart de zéro, clé de scellement
                 -- comprise. Sans cette ligne, une clé morte survivrait à
                 -- l'appairage qui devait tout remettre à plat, et TOUT
                 -- partirait en refus sur un maillage qui a l'air appairé.
                 seal_public_key=NULL,
                 seal_seen_at_ms=NULL""",
            (
                device_id,
                key,
                name,
                platform,
                device_type,
                TRUST_TRUSTED,
                json.dumps(sorted({str(c) for c in (declared_capabilities or [])})),
                str(app_version or "")[:40],
                stamp,
                stamp,
            ),
        )
        conn.commit()
    return self.get(device_id)

enrol_host

enrol_host(
    *,
    device_id: str,
    public_key_b64: str,
    name: str,
    platform: str = "UNKNOWN",
    device_type: str = "DESKTOP",
    declared_capabilities: Sequence[str] | None = None,
    address: str = "",
) -> dict[str, Any]

Enregistrer l'hôte QUI VIENT DE NOUS ACCUEILLIR dans la flotte.

Le pendant invité de redeem_pairing (Spatial Mesh, 25 août 2026) : sans lui, un Diapason qui rejoint sait parler à son hôte mais ne sait pas le reconnaître quand il répond — le jumelage n'était mutuel que d'un côté.

La confiance vient d'un fait, pas d'une déclaration : cette méthode n'est appelable qu'après avoir consommé AVEC SUCCÈS une invitation que l'hôte a lui-même émise. Les mêmes refus qu'au jumelage s'appliquent — une révocation ne se blanchit pas, et une clé qui change demande l'arbitrage de l'utilisateur.

Source code in src/diapason/mesh/registry.py

def enrol_host(
    self,
    *,
    device_id: str,
    public_key_b64: str,
    name: str,
    platform: str = "UNKNOWN",
    device_type: str = "DESKTOP",
    declared_capabilities: Sequence[str] | None = None,
    address: str = "",
) -> dict[str, Any]:
    """Enregistrer l'hôte QUI VIENT DE NOUS ACCUEILLIR dans la flotte.

    Le pendant invité de ``redeem_pairing`` (Spatial Mesh, 25 août
    2026) : sans lui, un Diapason qui rejoint sait parler à son hôte
    mais ne sait pas le reconnaître quand il répond — le jumelage
    n'était mutuel que d'un côté.

    La confiance vient d'un fait, pas d'une déclaration : cette méthode
    n'est appelable qu'après avoir consommé AVEC SUCCÈS une invitation
    que l'hôte a lui-même émise. Les mêmes refus qu'au jumelage
    s'appliquent — une révocation ne se blanchit pas, et une clé qui
    change demande l'arbitrage de l'utilisateur.
    """
    key = self._validate_public_key(public_key_b64)
    # La MÊME validation que redeem_pairing, et pas une plus sévère :
    # constaté le 25 août 2026 en jumelant deux vraies instances, un
    # hôte peut porter un identifiant HÉRITÉ (« mac-… », lu de la table
    # succes_meta) au lieu du « dev_… » dérivé de sa clé. Exiger le
    # préfixe ici rejetait la machine de développement elle-même — un
    # test avec un identifiant fabriqué ne l'aurait jamais montré.
    device_id = _clean(device_id, field="L'identifiant d'appareil", maximum=120)
    stamp = now_ms()
    with self._connect() as conn:
        existing = conn.execute(
            "SELECT public_key, trust_level FROM mesh_devices WHERE device_id=?",
            (device_id,),
        ).fetchone()
        if existing is not None:
            if existing["trust_level"] == TRUST_REVOKED:
                raise MeshError(
                    "Cet hôte a été révoqué ici. Supprimez-le de la liste "
                    "des appareils avant de le rejoindre à nouveau."
                )
            if existing["public_key"] != key:
                raise MeshError(
                    "Un appareil portant cet identifiant est déjà connu "
                    "avec une autre clé. Révoquez-le d'abord."
                )
        conn.execute(
            """INSERT INTO mesh_devices
               (device_id, public_key, name, platform, device_type,
                trust_level, declared_capabilities, app_version,
                created_at_ms, last_seen_at_ms)
               VALUES (?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(device_id) DO UPDATE SET
                 name=excluded.name,
                 platform=excluded.platform,
                 device_type=excluded.device_type,
                 trust_level=excluded.trust_level,
                 declared_capabilities=excluded.declared_capabilities,
                 last_seen_at_ms=excluded.last_seen_at_ms,
                 -- Même raison que pour l'autre porte d'appairage.
                 seal_public_key=NULL,
                 seal_seen_at_ms=NULL""",
            (
                device_id,
                key,
                str(name or "Hôte")[:80],
                str(platform or "UNKNOWN"),
                str(device_type or "DESKTOP"),
                TRUST_TRUSTED,
                json.dumps(sorted({str(c) for c in (declared_capabilities or [])})),
                "",
                stamp,
                stamp,
            ),
        )
        conn.commit()
    if address:
        try:
            return self.heartbeat(device_id, transport="lan", address=address)
        except MeshError:  # noqa: BLE001 - une adresse fausse n'annule pas le jumelage
            pass
    return self.get(device_id)

public_key_of

public_key_of(device_id: str) -> bytes | None

The key a command from this device must verify against.

Returns None for unknown OR revoked devices, so a caller cannot accidentally verify a signature from a device that was cut off.

Source code in src/diapason/mesh/registry.py

def public_key_of(self, device_id: str) -> bytes | None:
    """The key a command from this device must verify against.

    Returns None for unknown OR revoked devices, so a caller cannot
    accidentally verify a signature from a device that was cut off.
    """
    with closing(self._connect()) as conn:
        row = conn.execute(
            "SELECT public_key, trust_level FROM mesh_devices WHERE device_id=?",
            (device_id,),
        ).fetchone()
    if row is None or row["trust_level"] != TRUST_TRUSTED:
        return None
    return base64.b64decode(row["public_key"])

record_seal_key

record_seal_key(
    device_id: str, key_b64: str, sent_at_ms: int
) -> bool

Enregistrer la clé de scellement publiée par un pair.

Rend False quand la publication n'est pas STRICTEMENT plus récente que la dernière retenue — ce à quoi ressemble un rejeu. Le contrôle et l'écriture sont un seul UPDATE, exactement comme heartbeat_signed et pour la même raison : deux publications arrivant ensemble ne peuvent pas voir toutes deux l'ancienne marque, donc une seule gagne.

Un attaquant qui rejoue une vieille publication ne peut donc pas réinstaller une clé périmée dont il aurait, lui, la moitié privée.

Source code in src/diapason/mesh/registry.py

def record_seal_key(self, device_id: str, key_b64: str, sent_at_ms: int) -> bool:
    """Enregistrer la clé de scellement publiée par un pair.

    Rend False quand la publication n'est pas STRICTEMENT plus récente
    que la dernière retenue — ce à quoi ressemble un rejeu. Le contrôle et
    l'écriture sont un seul UPDATE, exactement comme ``heartbeat_signed``
    et pour la même raison : deux publications arrivant ensemble ne
    peuvent pas voir toutes deux l'ancienne marque, donc une seule gagne.

    Un attaquant qui rejoue une vieille publication ne peut donc pas
    réinstaller une clé périmée dont il aurait, lui, la moitié privée.
    """
    propre = str(key_b64 or "").strip()
    try:
        if len(base64.b64decode(propre, validate=True)) != 32:
            return False
    except Exception:  # noqa: BLE001 - une clé illisible n'est pas une clé
        return False

    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            "UPDATE mesh_devices SET seal_public_key=?, seal_seen_at_ms=? "
            "WHERE device_id=? AND trust_level=? "
            "  AND (seal_seen_at_ms IS NULL OR seal_seen_at_ms < ?)",
            (propre, int(sent_at_ms), device_id, TRUST_TRUSTED, int(sent_at_ms)),
        )
        conn.commit()
    return cursor.rowcount > 0

seal_key_of

seal_key_of(device_id: str) -> tuple[str, int] | None

La clé vers laquelle sceller, et QUAND elle a été vue.

Rend None pour un appareil inconnu, révoqué, ou qui n'a jamais publié — comme public_key_of, et pour la même raison : la révocation doit arrêter un appareil au même goulot que tout le reste.

L'instant est rendu avec la clé parce que l'appelant en a besoin : une clé trop ancienne ne s'emploie plus, et c'est le SEUL mécanisme de repli — on ne démote jamais sur un corps de réponse.

Source code in src/diapason/mesh/registry.py

def seal_key_of(self, device_id: str) -> tuple[str, int] | None:
    """La clé vers laquelle sceller, et QUAND elle a été vue.

    Rend None pour un appareil inconnu, révoqué, ou qui n'a jamais publié
    — comme ``public_key_of``, et pour la même raison : la révocation doit
    arrêter un appareil au même goulot que tout le reste.

    L'instant est rendu avec la clé parce que l'appelant en a besoin : une
    clé trop ancienne ne s'emploie plus, et c'est le SEUL mécanisme de
    repli — on ne démote jamais sur un corps de réponse.
    """
    with closing(self._connect()) as conn:
        row = conn.execute(
            "SELECT seal_public_key, seal_seen_at_ms, trust_level "
            "FROM mesh_devices WHERE device_id=?",
            (device_id,),
        ).fetchone()
    if row is None or row["trust_level"] != TRUST_TRUSTED:
        return None
    cle = row["seal_public_key"]
    vue = row["seal_seen_at_ms"]
    if not cle or vue is None:
        return None
    return str(cle), int(vue)

forget_seal_key

forget_seal_key(device_id: str) -> None

Oublier la clé d'un pair, pour repartir en clair immédiatement.

Le repli normal est l'expiration, qui prend sept jours. Ceci est la sortie de secours quand on sait déjà que le pair ne sait plus ouvrir ce qu'on lui scelle — une réinstallation, un retour en arrière — et qu'on ne veut pas attendre.

Source code in src/diapason/mesh/registry.py

def forget_seal_key(self, device_id: str) -> None:
    """Oublier la clé d'un pair, pour repartir en clair immédiatement.

    Le repli normal est l'expiration, qui prend sept jours. Ceci est la
    sortie de secours quand on sait déjà que le pair ne sait plus ouvrir
    ce qu'on lui scelle — une réinstallation, un retour en arrière — et
    qu'on ne veut pas attendre.
    """
    with closing(self._connect()) as conn, conn:
        conn.execute(
            "UPDATE mesh_devices SET seal_public_key=NULL, seal_seen_at_ms=NULL "
            "WHERE device_id=?",
            (device_id,),
        )
        conn.commit()

heartbeat

heartbeat(
    device_id: str,
    *,
    app_state: str = "",
    transport: str = "",
    address: str = "",
) -> dict[str, Any]

Record that a TRUSTED device is alive right now.

Restricted to trusted devices on purpose: a revoked one must not be able to make itself look reachable again simply by keeping a timer running.

Source code in src/diapason/mesh/registry.py

def heartbeat(
    self,
    device_id: str,
    *,
    app_state: str = "",
    transport: str = "",
    address: str = "",
) -> dict[str, Any]:
    """Record that a TRUSTED device is alive right now.

    Restricted to trusted devices on purpose: a revoked one must not be
    able to make itself look reachable again simply by keeping a timer
    running.
    """
    # The address is learned from the heartbeat rather than fixed at
    # pairing: a laptop changes network, and a stale address is worse
    # than none — it sends commands into the void.
    clean_address = str(address or "").strip()[:200]
    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            "UPDATE mesh_devices SET last_seen_at_ms=?, app_state=?, "
            "transport=?, address=COALESCE(NULLIF(?,''), address) "
            "WHERE device_id=? AND trust_level=?",
            (
                now_ms(),
                str(app_state or "")[:40] or None,
                str(transport or "")[:40] or None,
                clean_address,
                device_id,
                TRUST_TRUSTED,
            ),
        )
        conn.commit()
    if cursor.rowcount == 0:
        raise MeshError("Cet appareil n'est pas autorisé à signaler sa présence.")
    return self.get(device_id)

heartbeat_signed

heartbeat_signed(
    device_id: str,
    *,
    app_state: str = "",
    transport: str = "",
    address: str = "",
    app_version: str = "",
    sent_at_ms: int,
) -> dict[str, Any] | None

Record presence claimed by the device itself, once per timestamp.

Returns None when the beacon is not strictly newer than the last one accepted — which is what a replay looks like. The comparison and the write are one UPDATE on purpose: two beacons arriving together cannot both see the old watermark, so only one can win.

Source code in src/diapason/mesh/registry.py

def heartbeat_signed(
    self,
    device_id: str,
    *,
    app_state: str = "",
    transport: str = "",
    address: str = "",
    app_version: str = "",
    sent_at_ms: int,
) -> dict[str, Any] | None:
    """Record presence claimed by the device itself, once per timestamp.

    Returns ``None`` when the beacon is not strictly newer than the last
    one accepted — which is what a replay looks like. The comparison and
    the write are one UPDATE on purpose: two beacons arriving together
    cannot both see the old watermark, so only one can win.
    """
    clean_address = str(address or "").strip()[:200]
    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            "UPDATE mesh_devices SET last_seen_at_ms=?, last_beacon_at_ms=?, "
            "app_state=?, transport=?, "
            "address=COALESCE(NULLIF(?,''), address), "
            "app_version=COALESCE(NULLIF(?,''), app_version) "
            "WHERE device_id=? AND trust_level=? "
            "  AND (last_beacon_at_ms IS NULL OR last_beacon_at_ms < ?)",
            (
                now_ms(),
                int(sent_at_ms),
                str(app_state or "")[:40] or None,
                str(transport or "")[:40] or None,
                clean_address,
                str(app_version or "")[:40],
                device_id,
                TRUST_TRUSTED,
                int(sent_at_ms),
            ),
        )
        conn.commit()
    if cursor.rowcount == 0:
        return None
    return self.get(device_id)

declare_capabilities

declare_capabilities(
    device_id: str, capabilities: Iterable[str]
) -> dict[str, Any]

Record what a device CLAIMS. What it gets is computed on read.

Source code in src/diapason/mesh/registry.py

def declare_capabilities(
    self, device_id: str, capabilities: Iterable[str]
) -> dict[str, Any]:
    """Record what a device CLAIMS. What it gets is computed on read."""
    payload = json.dumps(
        sorted({str(c).strip() for c in capabilities if str(c).strip()})
    )
    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            "UPDATE mesh_devices SET declared_capabilities=?, last_seen_at_ms=? "
            "WHERE device_id=? AND trust_level=?",
            (payload, now_ms(), device_id, TRUST_TRUSTED),
        )
        conn.commit()
    if cursor.rowcount == 0:
        raise MeshError("Cet appareil n'est pas autorisé à déclarer ses capacités.")
    return self.get(device_id)

revoke

revoke(device_id: str) -> dict[str, Any]

Cut a device off. Terminal until the user deletes it outright.

Source code in src/diapason/mesh/registry.py

def revoke(self, device_id: str) -> dict[str, Any]:
    """Cut a device off. Terminal until the user deletes it outright."""
    with closing(self._connect()) as conn, conn:
        cursor = conn.execute(
            "UPDATE mesh_devices SET trust_level=?, revoked_at_ms=? "
            "WHERE device_id=?",
            (TRUST_REVOKED, now_ms(), device_id),
        )
        conn.commit()
    if cursor.rowcount == 0:
        raise MeshError("Cet appareil n'est pas enregistré.")
    return self.get(device_id)

forget

forget(device_id: str) -> None

Delete a device outright — the only way back from revocation.

Source code in src/diapason/mesh/registry.py

def forget(self, device_id: str) -> None:
    """Delete a device outright — the only way back from revocation."""
    with closing(self._connect()) as conn, conn:
        conn.execute("DELETE FROM mesh_devices WHERE device_id=?", (device_id,))
        conn.commit()

Functions: