feat(connect): add device identity store and registration proof (#6267)

rustfs/Cargo.toml

@@ -244,6 +244,10 @@ rustfs-concurrency = { workspace = true }
 rustfs-scanner = { workspace = true }
 tempfile = { workspace = true }
 
+# Connect device identity: P-256 keys, PKCS#10 certificate requests, ES256 proofs.
+p256 = { version = "0.13.2", features = ["ecdsa", "pkcs8"] }
+rcgen = { workspace = true }
+
 # Async Runtime and Networking
 async-trait = { workspace = true }
 axum.workspace = true

rustfs/src/connect/identity.rs

@@ -0,0 +1,265 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Device key, certificate request, and registration proof of possession.
+//!
+//! The transcript and signature rules implemented here are frozen by
+//! `protocol/agent/v1/registration-proof.md` and by the golden fixtures under
+//! `protocol/agent/v1/fixtures/registration/`. Connect verifies what this
+//! module produces, so any divergence is a protocol break rather than a
+//! local behaviour change.
+
+use base64::Engine as _;
+use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD};
+use p256::ecdsa::signature::Signer as _;
+use p256::ecdsa::{Signature, SigningKey};
+use p256::pkcs8::{DecodePrivateKey as _, EncodePrivateKey as _};
+use sha2::{Digest as _, Sha256};
+use zeroize::Zeroizing;
+
+/// The 30 US-ASCII octets that open every registration transcript. Case is
+/// significant: a lowercase spelling is a different transcript, and the
+/// protocol publishes it as a reject vector so the two can never be confused.
+const REGISTRATION_DOMAIN: &[u8] = b"RUSTFS-CONNECT-REGISTRATION-V1";
+
+/// Separator between a field's decimal octet length and its value.
+const FIELD_SEPARATOR: u8 = b':';
+
+/// Terminator after the domain and after every field value, including the last.
+const FIELD_TERMINATOR: u8 = b'\n';
+
+/// The transcript binds exactly seven fields, always present, always in order.
+const FIELD_COUNT: usize = 7;
+
+/// The one algorithm this surface accepts. The enumeration is closed: an
+/// unrecognised value is refused rather than discarded.
+pub const PROOF_ALGORITHM: &str = "ES256";
+
+#[derive(Debug, thiserror::Error)]
+pub enum IdentityError {
+    /// A transcript field carried an octet the encoding cannot represent
+    /// unambiguously. The transcript is length-prefixed, so a newline inside a
+    /// value would still parse; it is refused because a caller that can place
+    /// one can shift the boundary a verifier reconstructs from its own row.
+    #[error("registration transcript field {field} is not printable US-ASCII without a line feed")]
+    UnencodableField { field: &'static str },
+
+    /// An expiry that predates the epoch cannot be spelled without a sign, and
+    /// the length rule admits no sign.
+    #[error("registration token expiry {expires_unix} is negative")]
+    NegativeExpiry { expires_unix: i64 },
+
+    #[error("device key is not a valid P-256 private key: {0}")]
+    MalformedKey(String),
+
+    #[error("failed to generate the device certificate request: {0}")]
+    CertificateRequest(String),
+}
+
+/// The canonical byte sequence a device signs, and its digest.
+///
+/// Built, never parsed: nothing reads a transcript back, so there is no such
+/// thing as a malformed one once it has been constructed.
+#[derive(Clone)]
+pub struct RegistrationTranscript {
+    bytes: Vec<u8>,
+}
+
+impl std::fmt::Debug for RegistrationTranscript {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // The transcript embeds the challenge nonce, which is disclosed to an
+        // operator exactly once beside the token secret and is deliberately
+        // never republished. Rendering the octets would put it into any log or
+        // panic message that formats a transcript, so only the length and the
+        // digest — both already public in the fixtures — are shown.
+        f.debug_struct("RegistrationTranscript")
+            .field("len", &self.bytes.len())
+            .field("sha256", &self.sha256_hex())
+            .finish()
+    }
+}
+
+impl RegistrationTranscript {
+    /// Assemble the transcript from the seven bound values.
+    ///
+    /// Five of them reach the device out of band with the token secret and are
+    /// never sent back, which is what stops a device choosing its own
+    /// transcript. They cross an operator-supplied boundary, so each one is
+    /// checked here rather than trusted.
+    pub fn build(
+        registration_token_uid: &str,
+        organization_uid: &str,
+        cluster_uid: &str,
+        request_id: &str,
+        challenge_nonce: &str,
+        expires_unix: i64,
+        certificate_request: &[u8],
+    ) -> Result<Self, IdentityError> {
+        if expires_unix < 0 {
+            return Err(IdentityError::NegativeExpiry { expires_unix });
+        }
+
+        let expiry = expires_unix.to_string();
+        let csr_digest = BASE64_URL_NO_PAD.encode(Sha256::digest(certificate_request));
+
+        let fields: [(&'static str, &str); FIELD_COUNT] = [
+            ("registrationTokenUid", registration_token_uid),
+            ("organizationUid", organization_uid),
+            ("clusterUid", cluster_uid),
+            ("requestId", request_id),
+            ("challengeNonce", challenge_nonce),
+            ("expiresUnix", &expiry),
+            ("certificateRequestSha256", &csr_digest),
+        ];
+
+        let mut bytes = Vec::with_capacity(REGISTRATION_DOMAIN.len() + 1 + 320);
+        bytes.extend_from_slice(REGISTRATION_DOMAIN);
+        bytes.push(FIELD_TERMINATOR);
+
+        for (name, value) in fields {
+            if !value.is_ascii() || value.as_bytes().contains(&FIELD_TERMINATOR) {
+                return Err(IdentityError::UnencodableField { field: name });
+            }
+            // The length is the octet count, and `is_ascii` above makes octets
+            // and characters the same count for these values.
+            bytes.extend_from_slice(value.len().to_string().as_bytes());
+            bytes.push(FIELD_SEPARATOR);
+            bytes.extend_from_slice(value.as_bytes());
+            bytes.push(FIELD_TERMINATOR);
+        }
+
+        Ok(Self { bytes })
+    }
+
+    /// The exact octets that are signed.
+    pub fn as_bytes(&self) -> &[u8] {
+        &self.bytes
+    }
+
+    /// SHA-256 over the transcript, as lowercase hex. Published beside the
+    /// canonical string in `transcript.json` so a producer can prove its
+    /// builder without performing any cryptography.
+    pub fn sha256_hex(&self) -> String {
+        let digest = Sha256::digest(&self.bytes);
+        digest.iter().fold(String::with_capacity(64), |mut out, byte| {
+            use std::fmt::Write as _;
+            let _ = write!(out, "{byte:02x}");
+            out
+        })
+    }
+}
+
+/// A proof of possession, in the shape the exchange body carries.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RegistrationProof {
+    pub algorithm: String,
+    /// 86 base64url characters, unpadded, decoding to a fixed-width 64 octet
+    /// `r || s`.
+    pub value: String,
+}
+
+/// A device's P-256 key and the operations that key authorises.
+///
+/// The private key never leaves this type: it is not exposed by a getter, not
+/// rendered by `Debug`, and not written anywhere except the sealed store.
+pub struct DeviceIdentity {
+    signing_key: SigningKey,
+}
+
+impl std::fmt::Debug for DeviceIdentity {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // A device identity is a private key. Rendering any part of it, even a
+        // fingerprint, puts key-derived material into logs and support bundles.
+        f.write_str("DeviceIdentity(<redacted>)")
+    }
+}
+
+impl DeviceIdentity {
+    /// Generate a fresh P-256 key.
+    pub fn generate() -> Self {
+        // p256 is pinned to rand_core 0.6 while the workspace `rand` is 0.10, so
+        // the RNG comes from p256's own re-export rather than the workspace one.
+        Self {
+            signing_key: SigningKey::random(&mut p256::elliptic_curve::rand_core::OsRng),
+        }
+    }
+
+    /// Load a key from its PKCS#8 DER encoding. A key that does not decode is
+    /// an error rather than a reason to mint a replacement: silently
+    /// regenerating would strand the certificate already issued for the old one.
+    pub fn from_pkcs8_der(der: &[u8]) -> Result<Self, IdentityError> {
+        SigningKey::from_pkcs8_der(der)
+            .map(|signing_key| Self { signing_key })
+            .map_err(|error| IdentityError::MalformedKey(error.to_string()))
+    }
+
+    /// Serialise the key for the sealed store. The result is wrapped so it is
+    /// wiped when the caller drops it.
+    pub fn to_pkcs8_der(&self) -> Result<Zeroizing<Vec<u8>>, IdentityError> {
+        self.signing_key
+            .to_pkcs8_der()
+            .map(|der| Zeroizing::new(der.as_bytes().to_vec()))
+            .map_err(|error| IdentityError::MalformedKey(error.to_string()))
+    }
+
+    /// Build the PKCS#10 certificate request Connect consumes.
+    ///
+    /// Connect reads the request for its SubjectPublicKeyInfo and its
+    /// self-signature and for nothing else: it assigns the device uid itself,
+    /// so the subject and SAN carried here name nothing Connect will honour.
+    pub fn certificate_request_der(&self) -> Result<Vec<u8>, IdentityError> {
+        let pkcs8 = self.to_pkcs8_der()?;
+        let key_pair =
+            rcgen::KeyPair::try_from(pkcs8.as_slice()).map_err(|error| IdentityError::CertificateRequest(error.to_string()))?;
+
+        let params = rcgen::CertificateParams::default();
+        let request = params
+            .serialize_request(&key_pair)
+            .map_err(|error| IdentityError::CertificateRequest(error.to_string()))?;
+
+        Ok(request.der().to_vec())
+    }
+
+    /// Standard padded base64 of the certificate request, as the body carries it.
+    pub fn certificate_request_base64(&self) -> Result<String, IdentityError> {
+        Ok(BASE64_STANDARD.encode(self.certificate_request_der()?))
+    }
+
+    /// Sign a transcript, producing the low-S fixed-width proof.
+    ///
+    /// ECDSA admits two valid spellings of every signature, and a proof with
+    /// two spellings is not an identity, so `s` is normalised into the lower
+    /// half of the group order before encoding.
+    pub fn sign_registration(&self, transcript: &RegistrationTranscript) -> RegistrationProof {
+        let signature: Signature = self.signing_key.sign(transcript.as_bytes());
+        let canonical = signature.normalize_s().unwrap_or(signature);
+
+        RegistrationProof {
+            algorithm: PROOF_ALGORITHM.to_string(),
+            value: BASE64_URL_NO_PAD.encode(canonical.to_bytes()),
+        }
+    }
+
+    /// The device public key, DER SubjectPublicKeyInfo.
+    pub fn public_key_der(&self) -> Vec<u8> {
+        use p256::pkcs8::EncodePublicKey as _;
+
+        self.signing_key
+            .verifying_key()
+            .to_public_key_der()
+            .expect("a P-256 verifying key always encodes as SubjectPublicKeyInfo")
+            .as_bytes()
+            .to_vec()
+    }
+}

rustfs/src/connect/identity_store.rs

@@ -0,0 +1,244 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! On-disk home of the device key.
+//!
+//! A device that loses its key loses the certificate issued for it and has to
+//! spend a fresh registration token to get back, so the store is written
+//! durably and published exactly once. It deliberately does not reuse
+//! `rustfs_kms`'s `durable_file`, which implements the same commit protocol
+//! for envelope keys but is `pub(crate)` to that crate and carries KMS error
+//! and failpoint types this path has no use for.
+
+use std::fs;
+use std::io;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use zeroize::Zeroizing;
+
+use super::identity::{DeviceIdentity, IdentityError};
+
+/// Name of the key file inside the store directory.
+const KEY_FILE: &str = "device.key";
+
+/// Owner read/write only. The key is the device's whole identity.
+#[cfg(unix)]
+const KEY_MODE: u32 = 0o600;
+
+/// Distinguishes the staging file of concurrent publishers. The process id
+/// alone is not enough: several threads of one process may initialise the same
+/// store, and a shared staging name would let them truncate each other's
+/// half-written key and then link the result into place.
+static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0);
+
+#[derive(Debug, thiserror::Error)]
+pub enum StoreError {
+    #[error("connect identity store I/O failed at {path}: {source}")]
+    Io {
+        path: PathBuf,
+        #[source]
+        source: io::Error,
+    },
+
+    /// The key file exists but does not decode. Fail closed: regenerating here
+    /// would silently abandon a device certificate that is still valid and
+    /// still trusted by the control plane.
+    #[error("connect device key at {path} is unreadable and was left untouched: {source}")]
+    Corrupt {
+        path: PathBuf,
+        #[source]
+        source: IdentityError,
+    },
+
+    /// The key file is present with permissions that expose it. Refused rather
+    /// than repaired, because a key that has been world-readable has to be
+    /// treated as disclosed and rotated, not quietly re-sealed.
+    #[cfg(unix)]
+    #[error("connect device key at {path} has mode {mode:o}, expected {expected:o}")]
+    Permissions { path: PathBuf, mode: u32, expected: u32 },
+
+    #[error(transparent)]
+    Identity(#[from] IdentityError),
+}
+
+/// A directory holding one device identity.
+#[derive(Clone, Debug)]
+pub struct IdentityStore {
+    directory: PathBuf,
+}
+
+impl IdentityStore {
+    pub fn new(directory: impl Into<PathBuf>) -> Self {
+        Self {
+            directory: directory.into(),
+        }
+    }
+
+    pub fn key_path(&self) -> PathBuf {
+        self.directory.join(KEY_FILE)
+    }
+
+    /// Return the stored identity, or `None` when this deployment has never
+    /// been enrolled. Reading never creates anything, so an unconfigured
+    /// server can ask without acquiring an identity as a side effect.
+    pub fn load(&self) -> Result<Option<DeviceIdentity>, StoreError> {
+        let path = self.key_path();
+
+        let der = match fs::read(&path) {
+            Ok(der) => Zeroizing::new(der),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
+            Err(source) => return Err(StoreError::Io { path, source }),
+        };
+
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt as _;
+
+            let metadata = fs::metadata(&path).map_err(|source| StoreError::Io {
+                path: path.clone(),
+                source,
+            })?;
+            let mode = metadata.permissions().mode() & 0o7777;
+            if mode != KEY_MODE {
+                return Err(StoreError::Permissions {
+                    path,
+                    mode,
+                    expected: KEY_MODE,
+                });
+            }
+        }
+
+        DeviceIdentity::from_pkcs8_der(&der)
+            .map(Some)
+            .map_err(|source| StoreError::Corrupt { path, source })
+    }
+
+    /// Return the stored identity, generating and publishing one the first
+    /// time. Concurrent callers converge on a single identity: publication is
+    /// a no-clobber link, and whoever loses the race discards its candidate
+    /// and reads the winner's.
+    pub fn load_or_create(&self) -> Result<DeviceIdentity, StoreError> {
+        if let Some(identity) = self.load()? {
+            return Ok(identity);
+        }
+
+        fs::create_dir_all(&self.directory).map_err(|source| StoreError::Io {
+            path: self.directory.clone(),
+            source,
+        })?;
+
+        let candidate = DeviceIdentity::generate();
+        let der = candidate.to_pkcs8_der()?;
+
+        match self.publish(&der) {
+            Ok(()) => Ok(candidate),
+            // Another process published first. Its key is the identity; ours
+            // was never written anywhere and simply goes out of scope.
+            Err(StoreError::Io { source, .. }) if source.kind() == io::ErrorKind::AlreadyExists => {
+                self.load()?.ok_or_else(|| StoreError::Io {
+                    path: self.key_path(),
+                    source: io::Error::new(
+                        io::ErrorKind::NotFound,
+                        "device key vanished immediately after another writer published it",
+                    ),
+                })
+            }
+            Err(error) => Err(error),
+        }
+    }
+
+    /// Write, seal, fsync, then link into place and fsync the directory. The
+    /// key is durable before it is reachable, and it is reachable only once.
+    fn publish(&self, der: &[u8]) -> Result<(), StoreError> {
+        use std::io::Write as _;
+
+        let final_path = self.key_path();
+        let temp_path = self.directory.join(format!(
+            "{KEY_FILE}.{}.{}.tmp",
+            std::process::id(),
+            STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed)
+        ));
+
+        let io_at = |path: &Path| {
+            let path = path.to_path_buf();
+            move |source| StoreError::Io { path, source }
+        };
+
+        let mut options = fs::OpenOptions::new();
+        options.write(true).create(true).truncate(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt as _;
+            options.mode(KEY_MODE);
+        }
+
+        let mut file = options.open(&temp_path).map_err(io_at(&temp_path))?;
+
+        let result = (|| -> Result<(), StoreError> {
+            file.write_all(der).map_err(io_at(&temp_path))?;
+
+            // The umask can only narrow the creation mode, so set and verify
+            // the exact mode before the bytes become durable.
+            #[cfg(unix)]
+            {
+                use std::os::unix::fs::PermissionsExt as _;
+
+                file.set_permissions(fs::Permissions::from_mode(KEY_MODE))
+                    .map_err(io_at(&temp_path))?;
+                let mode = file.metadata().map_err(io_at(&temp_path))?.permissions().mode() & 0o7777;
+                if mode != KEY_MODE {
+                    return Err(StoreError::Permissions {
+                        path: temp_path.clone(),
+                        mode,
+                        expected: KEY_MODE,
+                    });
+                }
+            }
+
+            file.sync_all().map_err(io_at(&temp_path))?;
+            Ok(())
+        })();
+
+        drop(file);
+
+        if let Err(error) = result {
+            let _ = fs::remove_file(&temp_path);
+            return Err(error);
+        }
+
+        // `hard_link` fails rather than replacing an existing key, which is
+        // what makes a retry return the original identity instead of minting
+        // a second one.
+        let published = fs::hard_link(&temp_path, &final_path);
+        let _ = fs::remove_file(&temp_path);
+        published.map_err(io_at(&final_path))?;
+
+        fsync_dir(&self.directory).map_err(io_at(&self.directory))?;
+
+        Ok(())
+    }
+}
+
+/// Fsync a directory so a freshly linked entry survives power loss. Directories
+/// cannot be opened for syncing on Windows, where this is a no-op.
+fn fsync_dir(dir: &Path) -> io::Result<()> {
+    #[cfg(unix)]
+    {
+        fs::File::open(dir)?.sync_all()?;
+    }
+    #[cfg(not(unix))]
+    let _ = dir;
+    Ok(())
+}

rustfs/src/connect/mod.rs

@@ -0,0 +1,32 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! RustFS Connect device identity.
+//!
+//! A cluster device proves possession of its own key when it exchanges a
+//! one-time registration token for a durable certificate. This module owns the
+//! device-side half of that exchange: the P-256 key, the PKCS#10 certificate
+//! request built from it, and the proof-of-possession signature over the
+//! canonical transcript frozen by
+//! `protocol/agent/v1/registration-proof.md`.
+//!
+//! Nothing here contacts the network or starts a task. A deployment that has
+//! not been enrolled into a Connect control plane never calls into it, so an
+//! unconfigured server generates no key and holds no identity.
+
+pub mod identity;
+pub mod identity_store;
+
+pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript};
+pub use identity_store::{IdentityStore, StoreError};

rustfs/src/lib.rs

@@ -80,6 +80,7 @@ pub(crate) mod bitrot_selftest;
 pub mod capacity;
 pub mod cluster_snapshot;
 pub mod config;
+pub mod connect;
 pub mod delete_tail_activity;
 pub mod diagnose;
 pub mod embedded;

rustfs/tests/agent_protocol_fixtures.rs

@@ -0,0 +1,119 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Conformance of this repository's copy of the Connect agent protocol fixtures.
+//!
+//! `fixture-sets.json` requires a byte-identical copy of every populated set,
+//! and Connect's `make protocol-compat` runs this test by name (the Makefile's
+//! `RUSTFS_CONSUMER_TESTS` default) after comparing the two trees. The
+//! comparison there proves the copies match; this proves the copy is internally
+//! consistent, so a fixture edited on this side is caught here even when
+//! Connect is not checked out.
+
+use std::fs;
+use std::path::PathBuf;
+
+use sha2::{Digest as _, Sha256};
+
+fn fixture_root() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures")
+}
+
+fn sha256_hex(bytes: &[u8]) -> String {
+    Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+/// The registry is closed at eight sets; a ninth is a protocol change, not a
+/// fixture change. Mirrors `EXPECTED_SETS` in Connect's checker.
+const EXPECTED_SETS: [&str; 8] = [
+    "auth",
+    "version",
+    "registration",
+    "heartbeat",
+    "inventory",
+    "offline-enrollment",
+    "bundle",
+    "redaction",
+];
+
+#[test]
+fn agent_protocol_fixtures_registry_is_the_frozen_eight_sets() {
+    let registry: serde_json::Value =
+        serde_json::from_slice(&fs::read(fixture_root().join("fixture-sets.json")).expect("read fixture-sets.json"))
+            .expect("fixture-sets.json parses");
+
+    let names: Vec<&str> = registry["sets"]
+        .as_array()
+        .expect("sets is an array")
+        .iter()
+        .map(|set| set["name"].as_str().expect("set has a name"))
+        .collect();
+
+    assert_eq!(names, EXPECTED_SETS, "the fixture registry must stay closed and ordered");
+    assert_eq!(
+        registry["consumerCopy"]["path"].as_str(),
+        Some("protocol/agent/v1/fixtures"),
+        "this copy lives at the path the registry declares"
+    );
+}
+
+#[test]
+fn agent_protocol_fixtures_match_their_manifests() {
+    let root = fixture_root();
+    let registry: serde_json::Value =
+        serde_json::from_slice(&fs::read(root.join("fixture-sets.json")).expect("read fixture-sets.json"))
+            .expect("fixture-sets.json parses");
+
+    let mut checked = 0usize;
+
+    for set in registry["sets"].as_array().expect("sets is an array") {
+        let name = set["name"].as_str().expect("set has a name");
+        let status = set["status"].as_str().expect("set has a status");
+
+        let set_dir = root.join(name);
+        if status == "reserved" {
+            assert!(!set_dir.exists(), "reserved fixture set '{name}' must hold no files yet");
+            continue;
+        }
+
+        let manifest = fs::read_to_string(set_dir.join("MANIFEST.sha256"))
+            .unwrap_or_else(|error| panic!("populated set '{name}' must carry a manifest: {error}"));
+
+        let mut listed = Vec::new();
+        for line in manifest.lines().filter(|line| !line.trim().is_empty()) {
+            let (digest, file) = line
+                .split_once("  ")
+                .unwrap_or_else(|| panic!("malformed manifest line in '{name}': {line}"));
+            listed.push(file.to_string());
+
+            let bytes = fs::read(set_dir.join(file))
+                .unwrap_or_else(|error| panic!("set '{name}' lists {file} which is missing: {error}"));
+            assert_eq!(sha256_hex(&bytes), digest, "set '{name}' file {file} does not match its manifest");
+            checked += 1;
+        }
+
+        // A file present but unlisted would travel unchecked, so the manifest
+        // has to be exhaustive rather than merely correct about what it names.
+        let mut present: Vec<String> = fs::read_dir(&set_dir)
+            .expect("read fixture set directory")
+            .map(|entry| entry.expect("read dir entry").file_name().to_string_lossy().into_owned())
+            .filter(|file| file != "MANIFEST.sha256")
+            .collect();
+        present.sort();
+        listed.sort();
+        assert_eq!(present, listed, "set '{name}' holds files its manifest does not list");
+    }
+
+    assert!(checked > 0, "no fixture files were verified");
+}

rustfs/tests/connect_identity.rs

@@ -0,0 +1,506 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Connect device identity: transcript conformance, key durability, and the
+//! properties the registration exchange depends on.
+
+use std::fs;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use base64::Engine as _;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_URL_NO_PAD;
+use rustfs::connect::identity::{DeviceIdentity, IdentityError, RegistrationTranscript};
+use rustfs::connect::identity_store::{IdentityStore, StoreError};
+
+fn transcript_fixture() -> serde_json::Value {
+    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/registration/transcript.json");
+    serde_json::from_slice(&fs::read(path).expect("read transcript.json")).expect("transcript.json parses")
+}
+
+fn accept_vectors() -> serde_json::Value {
+    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/registration/accept-vectors.json");
+    serde_json::from_slice(&fs::read(path).expect("read accept-vectors.json")).expect("accept-vectors.json parses")
+}
+
+/// Extract the SubjectPublicKeyInfo from a PKCS#10 request.
+///
+/// The protocol freezes the DER prefix of a P-256 SubjectPublicKeyInfo, and the
+/// key that follows it is a 65 octet uncompressed point, so the whole structure
+/// is a fixed 91 octets located by its prefix. This is a test reading a fixture,
+/// not a parser: Connect owns certificate request parsing.
+fn subject_public_key_info(csr_der: &[u8]) -> Vec<u8> {
+    let prefix = hex_to_bytes("3059301306072a8648ce3d020106082a8648ce3d030107034200");
+    let start = csr_der
+        .windows(prefix.len())
+        .position(|window| window == prefix)
+        .expect("certificate request carries a P-256 SubjectPublicKeyInfo");
+    csr_der[start..start + prefix.len() + 65].to_vec()
+}
+
+/// Rebuild each accept vector's transcript from the values a verifier holds.
+///
+/// This is the interoperability assertion the protocol asks a producer to make:
+/// the five hidden fields come from the token row, the two visible ones from the
+/// request, and the result must equal the transcript Connect published.
+#[test]
+fn transcript_reproduces_every_accept_vector() {
+    let vectors = accept_vectors();
+    let list = vectors["vectors"].as_array().expect("accept vectors are a list");
+    assert!(!list.is_empty(), "the accept vector set must not be empty");
+
+    for vector in list {
+        let name = vector["name"].as_str().unwrap_or("<unnamed>");
+        let token = &vector["tokenRecord"];
+        let request = &vector["request"];
+
+        let csr = base64::engine::general_purpose::STANDARD
+            .decode(
+                request["certificateRequest"]
+                    .as_str()
+                    .expect("vector carries a certificate request"),
+            )
+            .expect("certificate request is base64");
+
+        let transcript = RegistrationTranscript::build(
+            token["registrationTokenUid"].as_str().unwrap(),
+            token["organizationUid"].as_str().unwrap(),
+            token["clusterUid"].as_str().unwrap(),
+            request["requestId"].as_str().unwrap(),
+            token["challengeNonce"].as_str().unwrap(),
+            token["expiresUnix"].as_i64().unwrap(),
+            &csr,
+        )
+        .unwrap_or_else(|error| panic!("vector '{name}' must build: {error}"));
+
+        assert_eq!(
+            transcript.as_bytes(),
+            vector["serverTranscript"].as_str().unwrap().as_bytes(),
+            "vector '{name}' transcript must match octet for octet"
+        );
+        assert_eq!(
+            transcript.sha256_hex(),
+            vector["serverTranscriptSha256"].as_str().unwrap(),
+            "vector '{name}' transcript digest must match"
+        );
+    }
+}
+
+/// The published proofs were produced by the Connect-side implementation over
+/// keys this repository does not hold. Verifying them against a transcript this
+/// module rebuilt is the strongest available statement that the two
+/// implementations agree: a single wrong octet anywhere in the transcript makes
+/// real ECDSA verification fail.
+#[test]
+fn published_proofs_verify_over_locally_rebuilt_transcripts() {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let vectors = accept_vectors();
+    let mut verified = 0usize;
+
+    for vector in vectors["vectors"].as_array().expect("accept vectors are a list") {
+        let name = vector["name"].as_str().unwrap_or("<unnamed>");
+        if vector["expected"]["verifiesMathematically"].as_bool() != Some(true) {
+            continue;
+        }
+
+        let token = &vector["tokenRecord"];
+        let request = &vector["request"];
+        let csr = base64::engine::general_purpose::STANDARD
+            .decode(request["certificateRequest"].as_str().unwrap())
+            .expect("certificate request is base64");
+
+        let transcript = RegistrationTranscript::build(
+            token["registrationTokenUid"].as_str().unwrap(),
+            token["organizationUid"].as_str().unwrap(),
+            token["clusterUid"].as_str().unwrap(),
+            request["requestId"].as_str().unwrap(),
+            token["challengeNonce"].as_str().unwrap(),
+            token["expiresUnix"].as_i64().unwrap(),
+            &csr,
+        )
+        .expect("transcript builds");
+
+        let raw = BASE64_URL_NO_PAD
+            .decode(request["proof"]["value"].as_str().expect("vector carries a proof"))
+            .expect("proof decodes");
+        let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
+        assert!(
+            signature.normalize_s().is_none(),
+            "vector '{name}' publishes a proof that is already low-S"
+        );
+
+        let verifying =
+            <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&subject_public_key_info(&csr))
+                .expect("public key decodes");
+
+        verifying
+            .verify(transcript.as_bytes(), &signature)
+            .unwrap_or_else(|error| panic!("vector '{name}' proof must verify over the rebuilt transcript: {error}"));
+        verified += 1;
+    }
+
+    assert!(verified > 0, "no accept vector was cross-verified");
+}
+
+/// Drive the builder with the golden example's own inputs, using the accept
+/// vector whose certificate request produces the digest it publishes.
+fn transcript_from_fixture_inputs(csr_octets: &[u8]) -> Result<RegistrationTranscript, IdentityError> {
+    let fixture = transcript_fixture();
+    let inputs = &fixture["example"]["inputs"];
+
+    RegistrationTranscript::build(
+        inputs["registrationTokenUid"].as_str().unwrap(),
+        inputs["organizationUid"].as_str().unwrap(),
+        inputs["clusterUid"].as_str().unwrap(),
+        inputs["requestId"].as_str().unwrap(),
+        inputs["challengeNonce"].as_str().unwrap(),
+        inputs["expiresUnix"].as_i64().unwrap(),
+        csr_octets,
+    )
+}
+
+fn csr_octets_matching_golden_digest() -> Vec<u8> {
+    let want = transcript_fixture()["example"]["inputs"]["certificateRequestSha256"]
+        .as_str()
+        .unwrap()
+        .to_string();
+
+    for vector in accept_vectors()["vectors"].as_array().expect("accept vectors are a list") {
+        let Some(encoded) = vector["request"]["certificateRequest"].as_str() else {
+            continue;
+        };
+        let der = base64::engine::general_purpose::STANDARD
+            .decode(encoded)
+            .expect("certificate request is base64");
+        let digest = BASE64_URL_NO_PAD.encode(<sha2::Sha256 as sha2::Digest>::digest(&der));
+        if digest == want {
+            return der;
+        }
+    }
+
+    panic!("no accept vector carries the certificate request the golden example digests");
+}
+
+#[test]
+fn transcript_reproduces_the_golden_example_byte_for_byte() {
+    let fixture = transcript_fixture();
+    let example = &fixture["example"];
+
+    let transcript = transcript_from_fixture_inputs(&csr_octets_matching_golden_digest()).expect("golden inputs build");
+
+    assert_eq!(
+        transcript.as_bytes(),
+        example["canonicalTranscript"].as_str().unwrap().as_bytes(),
+        "the canonical transcript must match octet for octet"
+    );
+    assert_eq!(
+        transcript.as_bytes().len() as u64,
+        example["canonicalTranscriptLengthBytes"].as_u64().unwrap(),
+        "the transcript length is frozen"
+    );
+    assert_eq!(
+        transcript.sha256_hex(),
+        example["canonicalTranscriptSha256"].as_str().unwrap(),
+        "the transcript digest is frozen"
+    );
+}
+
+#[test]
+fn transcript_refuses_a_field_carrying_the_terminator() {
+    // A newline inside a value would move the boundary a verifier rebuilds
+    // from its own token row, which is the substitution the encoding exists to
+    // prevent. Length-prefixing alone would still parse it.
+    let error = RegistrationTranscript::build(
+        "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5",
+        "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70\n36:evil",
+        "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81",
+        "3f2a1c94-5b6d-4e8f-9a0b-1c2d3e4f5a6b",
+        "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0f",
+        1_787_228_100,
+        b"csr",
+    )
+    .expect_err("a field carrying 0x0a must be refused");
+
+    assert!(
+        matches!(error, IdentityError::UnencodableField { field } if field == "organizationUid"),
+        "unexpected error: {error}"
+    );
+}
+
+#[test]
+fn transcript_refuses_a_non_ascii_field() {
+    let error = RegistrationTranscript::build(
+        "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5",
+        "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70",
+        "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81",
+        "3f2a1c94-5b6d-4e8f-9a0b-1c2d3e4f5a6b",
+        // Multi-byte input would make the octet length and the character count
+        // disagree, which is the exact confusion the length rule forbids.
+        "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0é",
+        1_787_228_100,
+        b"csr",
+    )
+    .expect_err("a non-ASCII field must be refused");
+
+    assert!(
+        matches!(error, IdentityError::UnencodableField { field } if field == "challengeNonce"),
+        "unexpected error: {error}"
+    );
+}
+
+#[test]
+fn transcript_refuses_a_negative_expiry() {
+    let error = transcript_negative_expiry().expect_err("a negative expiry has no unsigned spelling");
+    assert!(
+        matches!(error, IdentityError::NegativeExpiry { expires_unix: -1 }),
+        "unexpected error: {error}"
+    );
+}
+
+fn transcript_negative_expiry() -> Result<RegistrationTranscript, IdentityError> {
+    RegistrationTranscript::build(
+        "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5",
+        "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70",
+        "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81",
+        "3f2a1c94-5b6d-4e8f-9a0b-1c2d3e4f5a6b",
+        "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0f",
+        -1,
+        b"csr",
+    )
+}
+
+#[test]
+fn proof_is_a_canonical_low_s_signature_that_verifies() {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let identity = DeviceIdentity::generate();
+    let csr = identity.certificate_request_der().expect("certificate request builds");
+    let transcript = transcript_from_fixture_inputs(&csr).expect("transcript builds");
+
+    let proof = identity.sign_registration(&transcript);
+    assert_eq!(proof.algorithm, "ES256");
+    assert_eq!(proof.value.len(), 86, "the transfer encoding is 86 unpadded base64url characters");
+    assert!(
+        proof
+            .value
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
+        "the proof must use the base64url alphabet with no padding"
+    );
+
+    let raw = BASE64_URL_NO_PAD.decode(&proof.value).expect("proof decodes");
+    assert_eq!(raw.len(), 64, "the signature is a fixed-width r || s");
+
+    let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
+    assert!(
+        signature.normalize_s().is_none(),
+        "s must already be in the lower half of the group order"
+    );
+
+    let spki = identity.public_key_der();
+    let verifying =
+        <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&spki).expect("public key decodes");
+    verifying
+        .verify(transcript.as_bytes(), &signature)
+        .expect("the proof must verify over the transcript octets");
+}
+
+#[test]
+fn proof_does_not_verify_over_a_different_transcript() {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let identity = DeviceIdentity::generate();
+    let csr = identity.certificate_request_der().expect("certificate request builds");
+    let transcript = transcript_from_fixture_inputs(&csr).expect("transcript builds");
+    let proof = identity.sign_registration(&transcript);
+
+    // A different certificate request is a different artifact and therefore a
+    // different transcript; this is the proof-of-possession binding itself.
+    let other = transcript_from_fixture_inputs(b"a different certificate request").expect("transcript builds");
+    assert_ne!(transcript.as_bytes(), other.as_bytes());
+
+    let raw = BASE64_URL_NO_PAD.decode(&proof.value).expect("proof decodes");
+    let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
+    let verifying = <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&identity.public_key_der())
+        .expect("public key decodes");
+
+    assert!(
+        verifying.verify(other.as_bytes(), &signature).is_err(),
+        "a proof must not carry over to another transcript"
+    );
+}
+
+#[test]
+fn certificate_request_presents_a_p256_key() {
+    let identity = DeviceIdentity::generate();
+    let der = identity.certificate_request_der().expect("certificate request builds");
+
+    // The prefix the protocol freezes for a P-256 SubjectPublicKeyInfo. Its
+    // presence proves the request carries the curve Connect requires.
+    let spki_prefix = hex_to_bytes("3059301306072a8648ce3d020106082a8648ce3d030107034200");
+    assert!(
+        der.windows(spki_prefix.len()).any(|window| window == spki_prefix),
+        "the certificate request must present an ECDSA P-256 SubjectPublicKeyInfo"
+    );
+    assert_eq!(der[0], 0x30, "a PKCS#10 request is a DER SEQUENCE");
+}
+
+fn hex_to_bytes(hex: &str) -> Vec<u8> {
+    (0..hex.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("valid hex"))
+        .collect()
+}
+
+#[test]
+fn unenrolled_deployment_holds_no_identity_and_reading_creates_none() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let store = IdentityStore::new(dir.path().join("connect"));
+
+    assert!(store.load().expect("load succeeds").is_none(), "an unenrolled server has no identity");
+    assert!(
+        !dir.path().join("connect").exists(),
+        "reading must not create the store directory, let alone a key"
+    );
+}
+
+#[test]
+fn identity_survives_restart_and_retry_does_not_mint_a_second() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let store = IdentityStore::new(dir.path());
+
+    let first = store.load_or_create().expect("first create");
+    let first_key = first.public_key_der();
+
+    // A restart is a fresh store over the same directory.
+    let reopened = IdentityStore::new(dir.path());
+    let second = reopened.load_or_create().expect("second create");
+
+    assert_eq!(first_key, second.public_key_der(), "a retry must return the original identity");
+}
+
+#[test]
+fn concurrent_initialisation_converges_on_one_identity() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let path = dir.path().to_path_buf();
+    let started = Arc::new(AtomicUsize::new(0));
+
+    // Every thread must be spawned before any is joined: the barrier below
+    // makes each one wait for all eight, so joining as we spawn would both
+    // serialise the race this test exists to create and deadlock on the first
+    // thread. A lazy iterator chain here is not equivalent.
+    let mut handles = Vec::with_capacity(8);
+    for _ in 0..8 {
+        let path = path.clone();
+        let started = Arc::clone(&started);
+        handles.push(std::thread::spawn(move || {
+            // Line the threads up so publication actually races.
+            started.fetch_add(1, Ordering::SeqCst);
+            while started.load(Ordering::SeqCst) < 8 {
+                std::hint::spin_loop();
+            }
+            IdentityStore::new(&path).load_or_create().expect("create").public_key_der()
+        }));
+    }
+
+    let keys: Vec<Vec<u8>> = handles.into_iter().map(|handle| handle.join().expect("thread")).collect();
+
+    assert!(
+        keys.windows(2).all(|pair| pair[0] == pair[1]),
+        "every concurrent initialiser must observe the same device identity"
+    );
+}
+
+#[test]
+fn corrupt_key_is_refused_and_left_on_disk() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let store = IdentityStore::new(dir.path());
+    store.load_or_create().expect("create");
+
+    let key_path = store.key_path();
+    fs::write(&key_path, b"not a pkcs8 key").expect("corrupt the key");
+    set_mode(&key_path, 0o600);
+
+    let error = store.load().expect_err("a corrupt key must fail closed");
+    assert!(matches!(error, StoreError::Corrupt { .. }), "unexpected error: {error}");
+
+    // Regenerating would strand a certificate the control plane still trusts,
+    // so the damaged file has to survive for an operator to inspect.
+    assert_eq!(fs::read(&key_path).expect("key still present"), b"not a pkcs8 key");
+}
+
+#[cfg(unix)]
+#[test]
+fn key_is_sealed_and_widened_permissions_are_refused() {
+    use std::os::unix::fs::PermissionsExt as _;
+
+    let dir = tempfile::tempdir().expect("temp dir");
+    let store = IdentityStore::new(dir.path());
+    store.load_or_create().expect("create");
+
+    let key_path = store.key_path();
+    let mode = fs::metadata(&key_path).expect("metadata").permissions().mode() & 0o7777;
+    assert_eq!(mode, 0o600, "the device key must be owner-only");
+
+    set_mode(&key_path, 0o644);
+    let error = store.load().expect_err("a world-readable key must be refused");
+    assert!(matches!(error, StoreError::Permissions { mode: 0o644, .. }), "unexpected error: {error}");
+}
+
+#[cfg(unix)]
+fn set_mode(path: &std::path::Path, mode: u32) {
+    use std::os::unix::fs::PermissionsExt as _;
+    fs::set_permissions(path, fs::Permissions::from_mode(mode)).expect("set mode");
+}
+
+#[cfg(not(unix))]
+fn set_mode(_path: &std::path::Path, _mode: u32) {}
+
+#[test]
+fn unwritable_directory_fails_closed_without_publishing() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    let store_dir = dir.path().join("sealed");
+    fs::create_dir(&store_dir).expect("create store dir");
+    set_mode(&store_dir, 0o500);
+
+    let store = IdentityStore::new(&store_dir);
+    let result = store.load_or_create();
+
+    set_mode(&store_dir, 0o700);
+
+    #[cfg(unix)]
+    {
+        assert!(result.is_err(), "an unwritable store must not silently succeed");
+        assert!(!store.key_path().exists(), "no key may be published when the write failed");
+    }
+    #[cfg(not(unix))]
+    let _ = result;
+}
+
+#[test]
+fn stored_key_round_trips_through_pkcs8() {
+    let identity = DeviceIdentity::generate();
+    let der = identity.to_pkcs8_der().expect("serialise");
+    let reloaded = DeviceIdentity::from_pkcs8_der(&der).expect("deserialise");
+
+    assert_eq!(identity.public_key_der(), reloaded.public_key_der(), "the key must survive a round trip");
+}
+
+#[test]
+fn device_identity_does_not_render_key_material() {
+    let identity = DeviceIdentity::generate();
+    assert_eq!(format!("{identity:?}"), "DeviceIdentity(<redacted>)");
+}