From ab5d433fe1794546456255eb25d67f84961ad11e Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 22 Aug 2026 07:07:29 +0800 Subject: [PATCH] feat(connect): add registration and credential rotation (#6353) --- Cargo.lock | 1 + Cargo.toml | 1 + .../agent/v1/fixtures/auth/MANIFEST.sha256 | 5 +- .../agent/v1/fixtures/auth/error-codes.json | 20 +- .../v1/fixtures/auth/rotation-proof.json | 306 ++++++ .../v1/fixtures/auth/surface-separation.json | 2 +- protocol/agent/v1/fixtures/fixture-sets.json | 2 +- rustfs/Cargo.toml | 1 + rustfs/src/connect/client.rs | 566 +++++++++++ rustfs/src/connect/credential_store.rs | 370 +++++++ rustfs/src/connect/identity.rs | 8 +- rustfs/src/connect/identity_store.rs | 131 ++- rustfs/src/connect/mod.rs | 6 + rustfs/src/connect/registration.rs | 527 ++++++++++ rustfs/tests/connect_registration.rs | 911 ++++++++++++++++++ 15 files changed, 2828 insertions(+), 29 deletions(-) create mode 100644 protocol/agent/v1/fixtures/auth/rotation-proof.json create mode 100644 rustfs/src/connect/client.rs create mode 100644 rustfs/src/connect/credential_store.rs create mode 100644 rustfs/src/connect/registration.rs create mode 100644 rustfs/tests/connect_registration.rs diff --git a/Cargo.lock b/Cargo.lock index 312fc1376..2cb35c808 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9257,6 +9257,7 @@ dependencies = [ "url", "urlencoding", "uuid", + "x509-parser", "zeroize", "zip", "zstd", diff --git a/Cargo.toml b/Cargo.toml index 652e931e2..918af3cce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -204,6 +204,7 @@ rsa = { version = "=0.10.0-rc.18" } rustls = { default-features = false, version = "0.23.43" } rustls-native-certs = "0.8" rustls-pki-types = "1.15.1" +x509-parser = "0.18.1" sha1 = "0.11.0" sha2 = "0.11.0" subtle = "2.6" diff --git a/protocol/agent/v1/fixtures/auth/MANIFEST.sha256 b/protocol/agent/v1/fixtures/auth/MANIFEST.sha256 index f43d5f192..6df7052ef 100644 --- a/protocol/agent/v1/fixtures/auth/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/auth/MANIFEST.sha256 @@ -1,5 +1,6 @@ 3d602080f7ca4c32ba9e37ad1a32665c78560726b30aeee08fd9e95eb2f36194 accept-vectors.json d3c19946288717088145592e0e8d6f2fa684443ba2f73d4c7bc49c415d6dd051 certificate-profile.json -060485263c51003274c056a0e04bec1b7d76157cf599ba79eebe040bc7cee71b error-codes.json +299a2ae34a8ca74bcf31deeb53a08f9eff279efa09d3f5358a0cf10866fe1a5d error-codes.json 43fe297ffb512b1b9f4af62f1832f3aa3905157893bfdc3dcc6d56f5a98aaef6 reject-vectors.json -b946175b094f4a8d75091b652fbe3d4327c9c795f28e02c96e1ab90a429e418d surface-separation.json +0cf26a7332fa6e3f57390e081f2cceead3236f6ca57f7038e3b55c2582af1733 rotation-proof.json +7c07100460fa23fca482466df9ca22f91c7f26b36087a7207a001f7d987f527e surface-separation.json diff --git a/protocol/agent/v1/fixtures/auth/error-codes.json b/protocol/agent/v1/fixtures/auth/error-codes.json index 6a934b139..1af792add 100644 --- a/protocol/agent/v1/fixtures/auth/error-codes.json +++ b/protocol/agent/v1/fixtures/auth/error-codes.json @@ -2,7 +2,7 @@ "protocolVersion": "v1", "fixtureSet": "auth", "fixture": "error-codes", - "description": "Frozen ErrorInfo reasons for agent authentication and negotiation. Clients branch on status and reason, never on message.", + "description": "Frozen ErrorInfo reasons for agent authentication, negotiation, and credential-rotation authorization. Clients branch on status and reason, never on message.", "domain": "rustfs.connect", "detailType": "type.googleapis.com/google.rpc.ErrorInfo", "disclosureRules": [ @@ -81,6 +81,24 @@ "httpStatus": 401, "status": "UNAUTHENTICATED", "meaning": "A browser session cookie was presented to an authenticated agent operation. The agent surface never accepts it." + }, + { + "reason": "ROTATION_CREDENTIAL_NOT_CURRENT", + "httpStatus": 409, + "status": "ABORTED", + "meaning": "The authenticated certificate is valid for ordinary agent operations but is not the device's current ACTIVE credential and therefore cannot authorize another rotation." + }, + { + "reason": "ROTATION_REQUEST_CONFLICT", + "httpStatus": 409, + "status": "ABORTED", + "meaning": "The requestId already belongs to a rotation with different transcript inputs and cannot be reused." + }, + { + "reason": "ROTATION_PROOF_INVALID", + "httpStatus": 401, + "status": "UNAUTHENTICATED", + "meaning": "The credential proof does not verify over the canonical rotation transcript under the presented certificate key." } ] } diff --git a/protocol/agent/v1/fixtures/auth/rotation-proof.json b/protocol/agent/v1/fixtures/auth/rotation-proof.json new file mode 100644 index 000000000..a65b83a30 --- /dev/null +++ b/protocol/agent/v1/fixtures/auth/rotation-proof.json @@ -0,0 +1,306 @@ +{ + "protocolVersion": "v1", + "fixtureSet": "auth", + "fixture": "rotation-proof", + "description": "Frozen proof-of-possession transcript for rotating an online device credential. The current certificate key authorizes one new certificate request for one device and one idempotent request.", + "operation": { + "method": "POST", + "path": "/agent/clusterDevices/{device}:rotateCredential", + "operationId": "rotateClusterDeviceCredential", + "authenticatedSurface": "/agent/*", + "currentCredentialRequiredForNewRotation": true, + "outgoingOverlapCredentialMayAuthenticateOrdinaryOperations": true, + "outgoingOverlapCredentialMayRotate": false, + "outgoingOverlapCredentialMayReplayItsCompletedRotation": true + }, + "replayPolicy": { + "recordState": "COMPLETED", + "bindingFields": [ + "currentCertificateFingerprint", + "clusterDeviceName", + "requestId", + "certificateRequestSha256" + ], + "exactMatch": "returnStoredResult", + "mismatchReason": "ROTATION_REQUEST_CONFLICT", + "retentionLowerBound": "outgoingCredential.validUntil", + "sideEffects": "No issuer call, credential write, or overlap extension." + }, + "completedReplayRecord": { + "state": "COMPLETED", + "currentCertificateFingerprint": "1bc816e2d285c52ee3a0aa06dc7aec17e788626043e795259a67f2436f970d09", + "clusterDeviceName": "organizations/0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92", + "requestId": "7c5d6e2a-8b91-4f03-a4d5-6e7f8091a2b3", + "certificateRequestSha256": "kVS-bXYxD6F22cZNy4Vnpgb5gFZbr4EGP5GvIz7uOaw", + "resultReference": "credential-rotation-result-01" + }, + "transcript": { + "domain": "RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1", + "domainTerminator": "0x0a", + "fieldSeparator": "0x3a", + "fieldTerminator": "0x0a", + "fieldCount": 4, + "encoding": "US-ASCII", + "normalisationPermitted": false, + "fieldOrder": [ + "currentCertificateFingerprint", + "clusterDeviceName", + "requestId", + "certificateRequestSha256" + ], + "fields": [ + { + "name": "currentCertificateFingerprint", + "position": 1, + "source": "the exact certificate accepted by trusted ingress and resolved by Connect", + "pattern": "^[0-9a-f]{64}$", + "binds": "the one currently active credential authorizing the rotation", + "absenceWouldAllow": "A proof captured from one current certificate to authorize another credential after the device rotated." + }, + { + "name": "clusterDeviceName", + "position": 2, + "source": "the authenticated identity after it is matched to the target resource", + "pattern": "^organizations/[0-9a-f-]{36}/clusters/[0-9a-f-]{36}/clusterDevices/[0-9a-f-]{36}$", + "binds": "the organization, cluster, and device being rotated", + "absenceWouldAllow": "A proof produced by one device to be presented against another device resource." + }, + { + "name": "requestId", + "position": 3, + "source": "the request body", + "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", + "binds": "the single idempotent rotation attempt", + "absenceWouldAllow": "A captured proof to be replayed as a fresh attempt instead of the same idempotent request." + }, + { + "name": "certificateRequestSha256", + "position": 4, + "source": "recomputed over the exact PKCS#10 DER octets in certificateRequest", + "pattern": "^[A-Za-z0-9_-]{43}$", + "binds": "the next device public key and certificate request", + "absenceWouldAllow": "A valid proof to be combined with an attacker certificate request." + } + ] + }, + "signature": { + "algorithmField": "proof.algorithm", + "algorithmEnumeration": ["ES256"], + "curve": "P-256", + "signatureEncoding": "fixed-width-r-s", + "signatureLengthBytes": 64, + "signatureTransferEncoding": "base64url-unpadded", + "signatureValuePattern": "^[A-Za-z0-9_-]{86}$", + "lowSRequired": true, + "verifyingKeySource": "the SubjectPublicKeyInfo of the exact presented certificate accepted by trusted ingress", + "newKeyPossession": "the PKCS#10 self-signature is verified separately under the public key inside certificateRequest", + "sharedEncodingContract": "protocol/agent/v1/registration-proof.md#the-signature" + }, + "verificationOrder": [ + { + "stage": "protocolVersion", + "rule": "protocolVersion must be the supported major version.", + "reason": "UNSUPPORTED_PROTOCOL" + }, + { + "stage": "encoding", + "rule": "proof.algorithm and proof.value obey the shared ES256 fixed-width low-S contract.", + "reasons": ["UNSUPPORTED_ALGORITHM", "SIGNATURE_MALFORMED", "SIGNATURE_NOT_CANONICAL"] + }, + { + "stage": "certificateRequest", + "rule": "certificateRequest is one self-signed PKCS#10 request whose public key is P-256.", + "reasons": ["CERTIFICATE_REQUEST_MALFORMED", "DEVICE_KEY_UNSUPPORTED"] + }, + { + "stage": "proof", + "rule": "Connect rebuilds the transcript and verifies proof.value under the presented certificate public key.", + "reason": "ROTATION_PROOF_INVALID" + }, + { + "stage": "replay", + "rule": "A completed record bound to the presented credential, device, requestId, and CSR digest returns its stored result without issuing again. A requestId bound to different transcript inputs is refused.", + "reason": "ROTATION_REQUEST_CONFLICT" + }, + { + "stage": "credential", + "rule": "After replay lookup misses, the authenticated certificate must be the device current ACTIVE credential, not a credential in the outgoing overlap.", + "reason": "ROTATION_CREDENTIAL_NOT_CURRENT" + } + ], + "reasonSources": { + "auth": ["UNSUPPORTED_PROTOCOL", "ROTATION_CREDENTIAL_NOT_CURRENT", "ROTATION_REQUEST_CONFLICT", "ROTATION_PROOF_INVALID"], + "sharedFromRegistration": [ + "UNSUPPORTED_ALGORITHM", + "SIGNATURE_MALFORMED", + "SIGNATURE_NOT_CANONICAL", + "CERTIFICATE_REQUEST_MALFORMED", + "DEVICE_KEY_UNSUPPORTED" + ] + }, + "example": { + "inputs": { + "currentCertificateFingerprint": "1bc816e2d285c52ee3a0aa06dc7aec17e788626043e795259a67f2436f970d09", + "clusterDeviceName": "organizations/0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92", + "requestId": "7c5d6e2a-8b91-4f03-a4d5-6e7f8091a2b3", + "certificateRequestSha256": "kVS-bXYxD6F22cZNy4Vnpgb5gFZbr4EGP5GvIz7uOaw" + }, + "artifacts": { + "currentPublicKeySpki": "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAENEpjuFZTqp0Hxh/OWV3TGkluNjCo15dk+4CozuR6aT9Vaxhkb2M9nhaVGfk8+aSiSIiKFSCsYonKl8jh743Qow==", + "currentPublicKeyFingerprint": "28608e223c75ed89e72041f12afae4fc1cd3f1d3bcca38cbb62d1a41687610c2", + "certificateRequest": "MIIBHjCBxAIBADBiMRswGQYDVQQDDBJpZ25vcmVkLWJ5LWNvbm5lY3QxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQWPosvIKr5V3CpTUdLfMQxDx31B3SglLKyqRg/oH3J+PhUnqf7pDWW1sTkP2aRIDjAwRn0DpLrz405CcvGHvYEoAAwCgYIKoZIzj0EAwIDSQAwRgIhAMJzXo/CK4E9BfjOxP35he9LLlqENhK7HTzZQTuIgLX2AiEAwOZHibk5HEijTWcJ/UT117nssfJesWZVOWwz/KTIpi8=", + "otherCertificateRequest": "MIIBHTCBxAIBADBiMRswGQYDVQQDDBJpZ25vcmVkLWJ5LWNvbm5lY3QxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQwWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAATi0KiZGcbPJixDiQN/aIVky7xqaYhjmwAn1aozrC2eanaDENYQ3D5zw5qoZNIp8YUk/CETZrba3C5KYR5ocdlLoAAwCgYIKoZIzj0EAwIDSAAwRQIhAL0ZtvWyTSu18RF5J4ZVIuOGjJpJwSdP+87CVxNKJJduAiAahIruc3FLtO3RI7B8Ome8IsVDUpSAQilThjdOeDMsYg==" + }, + "canonicalTranscript": "RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1\n64:1bc816e2d285c52ee3a0aa06dc7aec17e788626043e795259a67f2436f970d09\n148:organizations/0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92\n36:7c5d6e2a-8b91-4f03-a4d5-6e7f8091a2b3\n43:kVS-bXYxD6F22cZNy4Vnpgb5gFZbr4EGP5GvIz7uOaw\n", + "canonicalTranscriptLengthBytes": 346, + "canonicalTranscriptSha256": "e5f0e9cd0d5d420bc7e51217512de76e1cf4dceb1d67804b6518c4fb5d9fe434" + }, + "acceptVectors": [ + { + "name": "current credential authorizes one new certificate request", + "requestId": "7c5d6e2a-8b91-4f03-a4d5-6e7f8091a2b3", + "proof": "vgz7Hm-xZ5JQ7B3oqUEKXB4tQxEzWZ988iPxAklaIZIsd4nSTUv5YriqES5yKGHvKOgWPyWjvrgI0j0T-fYEsQ", + "expected": {"accepted": true, "reason": null, "stage": "credential", "replayed": false} + }, + { + "name": "the same certificate request under a new requestId needs its own proof", + "requestId": "82e1f3a4-9b05-4c67-8d90-1e2f3a4b5c6d", + "proof": "s-rkF9w41sqgdQNkOlhjcLYlrNHN_HL5xh4C3k_1a9ZQu7Uq0LoET_x7i4Rw3xQ75ec6B6tfqZ3XsHTS0FfN7Q", + "expected": {"accepted": true, "reason": null, "stage": "credential", "replayed": false} + }, + { + "name": "outgoing credential replays its completed byte-equivalent rotation", + "requestId": "7c5d6e2a-8b91-4f03-a4d5-6e7f8091a2b3", + "proof": "vgz7Hm-xZ5JQ7B3oqUEKXB4tQxEzWZ988iPxAklaIZIsd4nSTUv5YriqES5yKGHvKOgWPyWjvrgI0j0T-fYEsQ", + "replayRecord": "completedExample", + "mutation": {"currentCredential": false}, + "expected": {"accepted": true, "reason": null, "stage": "replay", "replayedResultReference": "credential-rotation-result-01"} + } + ], + "rejectVectors": [ + { + "name": "outgoing overlap credential attempts another rotation", + "stage": "credential", + "requestId": "82e1f3a4-9b05-4c67-8d90-1e2f3a4b5c6d", + "proof": "s-rkF9w41sqgdQNkOlhjcLYlrNHN_HL5xh4C3k_1a9ZQu7Uq0LoET_x7i4Rw3xQ75ec6B6tfqZ3XsHTS0FfN7Q", + "mutation": {"currentCredential": false}, + "expected": {"accepted": false, "reason": "ROTATION_CREDENTIAL_NOT_CURRENT"} + }, + { + "name": "rotation declares protocol v2", + "stage": "protocolVersion", + "mutation": {"protocolVersion": "v2"}, + "expected": {"accepted": false, "reason": "UNSUPPORTED_PROTOCOL"} + }, + { + "name": "proof declares ES384", + "stage": "encoding", + "mutation": {"proofAlgorithm": "ES384"}, + "expected": {"accepted": false, "reason": "UNSUPPORTED_ALGORITHM"} + }, + { + "name": "DER encoded proof", + "stage": "encoding", + "mutation": {"proofValue": "MEUCIQC-DPseb7FnklDsHeipQQpcHi1DETNZn3zyI_ECSVohkgIgLHeJ0k1L-WK4qhEucihh7yjoFj8lo764CNI9E_n2BLE"}, + "expected": {"accepted": false, "reason": "SIGNATURE_MALFORMED"} + }, + { + "name": "padded base64url proof", + "stage": "encoding", + "mutation": {"proofValue": "vgz7Hm-xZ5JQ7B3oqUEKXB4tQxEzWZ988iPxAklaIZIsd4nSTUv5YriqES5yKGHvKOgWPyWjvrgI0j0T-fYEsQ=="}, + "expected": {"accepted": false, "reason": "SIGNATURE_MALFORMED"} + }, + { + "name": "protocol negotiation precedes proof encoding and stale credential state", + "stage": "protocolVersion", + "mutation": {"currentCredential": false, "protocolVersion": "v2", "proofValue": "not-a-signature"}, + "expected": {"accepted": false, "reason": "UNSUPPORTED_PROTOCOL"} + }, + { + "name": "declared algorithm precedes signature bytes", + "stage": "encoding", + "mutation": {"proofAlgorithm": "ES384", "proofValue": "not-a-signature"}, + "expected": {"accepted": false, "reason": "UNSUPPORTED_ALGORITHM"} + }, + { + "name": "malleated high-S proof", + "stage": "encoding", + "mutation": {"proofValue": "vgz7Hm-xZ5JQ7B3oqUEKXB4tQxEzWZ988iPxAklaIZLTiHYssrQGnkdV7tGN154Qk_7kboFz38zq542vAm0goA"}, + "expected": {"accepted": false, "reason": "SIGNATURE_NOT_CANONICAL"} + }, + { + "name": "proof over the rotation fields uses the registration domain", + "stage": "proof", + "mutation": {"proofValue": "as27GKXEKXtym-BU8NUl0BYhJkEUooZWPadrxwOWa40pAaN6p4VxKCCFggZl4ZQsl5CUtaMoMaxM_HzT1c1qUA"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "proof verification precedes stale credential state", + "stage": "proof", + "mutation": {"currentCredential": false, "proofValue": "as27GKXEKXtym-BU8NUl0BYhJkEUooZWPadrxwOWa40pAaN6p4VxKCCFggZl4ZQsl5CUtaMoMaxM_HzT1c1qUA"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "new certificate request key signs instead of the current credential key", + "stage": "proof", + "mutation": {"proofValue": "QEN8wMQQk5HjriDw2aeAxIfsRvrWgIjGR3KlpK81Y34kaTAacO-6_bOMAjQ0hMMIk-YYN6MDCw0Via_Ri0X0BA"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "proof is moved to another current certificate", + "stage": "proof", + "mutation": {"currentCertificateFingerprint": "2bc816e2d285c52ee3a0aa06dc7aec17e788626043e795259a67f2436f970d09"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "proof is moved to another device", + "stage": "proof", + "mutation": {"clusterDeviceName": "organizations/0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e93"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "proof is replayed under another requestId", + "stage": "proof", + "mutation": {"requestId": "82e1f3a4-9b05-4c67-8d90-1e2f3a4b5c6d"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "proof is combined with another certificate request", + "stage": "proof", + "mutation": {"certificateRequest": "otherCertificateRequest"}, + "expected": {"accepted": false, "reason": "ROTATION_PROOF_INVALID"} + }, + { + "name": "completed replay belongs to another certificate fingerprint", + "stage": "replay", + "replayRecord": "completedExample", + "replayRecordMutation": {"currentCertificateFingerprint": "2bc816e2d285c52ee3a0aa06dc7aec17e788626043e795259a67f2436f970d09"}, + "mutation": {"currentCredential": false}, + "expected": {"accepted": false, "reason": "ROTATION_REQUEST_CONFLICT"} + }, + { + "name": "completed replay belongs to another device", + "stage": "replay", + "replayRecord": "completedExample", + "replayRecordMutation": {"clusterDeviceName": "organizations/0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70/clusters/0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81/clusterDevices/0198f4b0-3c00-7e30-8f41-4a5b6c7d8e93"}, + "mutation": {"currentCredential": false}, + "expected": {"accepted": false, "reason": "ROTATION_REQUEST_CONFLICT"} + }, + { + "name": "completed replay belongs to another requestId", + "stage": "replay", + "replayRecord": "completedExample", + "replayRecordMutation": {"requestId": "82e1f3a4-9b05-4c67-8d90-1e2f3a4b5c6d"}, + "mutation": {"currentCredential": false}, + "expected": {"accepted": false, "reason": "ROTATION_REQUEST_CONFLICT"} + }, + { + "name": "completed replay belongs to another CSR digest", + "stage": "replay", + "replayRecord": "completedExample", + "replayRecordMutation": {"certificateRequestSha256": "PXxfNGWtIbrqVocag6lOLDDR41AjpHXJ7UcAWkbTSAs"}, + "mutation": {"currentCredential": false}, + "expected": {"accepted": false, "reason": "ROTATION_REQUEST_CONFLICT"} + } + ] +} diff --git a/protocol/agent/v1/fixtures/auth/surface-separation.json b/protocol/agent/v1/fixtures/auth/surface-separation.json index 4d3f8a895..611a42cab 100644 --- a/protocol/agent/v1/fixtures/auth/surface-separation.json +++ b/protocol/agent/v1/fixtures/auth/surface-separation.json @@ -49,7 +49,7 @@ "schemeType": "mutualTLS", "forbiddenSecuritySchemes": ["sessionCookie"], "defaultSecurity": ["agentMutualTls"], - "publicOperations": ["getProtocolStatus"] + "publicOperations": ["getProtocolStatus", "exchangeRegistrationToken"] }, { "document": "openapi/control.json", diff --git a/protocol/agent/v1/fixtures/fixture-sets.json b/protocol/agent/v1/fixtures/fixture-sets.json index 5ded06740..d3beb8f23 100644 --- a/protocol/agent/v1/fixtures/fixture-sets.json +++ b/protocol/agent/v1/fixtures/fixture-sets.json @@ -11,7 +11,7 @@ { "name": "auth", "status": "populated", - "purpose": "Client certificate profile, RFC 9440 header profile, authentication accept and reject vectors, surface separation, and the frozen error reason registry." + "purpose": "Client certificate profile, RFC 9440 header profile, authentication and credential-rotation proof vectors, surface separation, and the frozen error reason registry." }, { "name": "version", diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 206d9dfcc..28f4ad259 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -288,6 +288,7 @@ zstd.workspace = true # Cryptography and Security rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] } rustls-pki-types = { workspace = true } +x509-parser = { workspace = true } subtle = { workspace = true } jiff = { workspace = true, features = ["serde"] } time = { workspace = true, features = ["parsing", "formatting", "serde", "macros"] } diff --git a/rustfs/src/connect/client.rs b/rustfs/src/connect/client.rs new file mode 100644 index 000000000..a13ec953d --- /dev/null +++ b/rustfs/src/connect/client.rs @@ -0,0 +1,566 @@ +// 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. + +use std::time::Duration; + +use base64::Engine as _; +use reqwest::{Client, StatusCode, Url}; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, pem::PemObject as _}; +use serde::Deserialize; +use uuid::Uuid; +use zeroize::Zeroizing; + +use super::credential_store::{ + CompletedRegistration, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration, PendingRotation, +}; +use super::identity::{IdentityError, RegistrationTranscript}; +use super::identity_store::{IdentityStore, StoreError}; +use super::registration::{ + CredentialResponse, CredentialValidationError, ExpectedDevice, RegistrationRequest, RegistrationToken, RotationRequest, + certificate_fingerprint, certificate_request_matches, public_key_fingerprint, validate_credential, + validate_stored_credential, +}; + +const MAX_ATTEMPTS: usize = 3; +const MAX_RESPONSE_BYTES: usize = 1024 * 1024; +const ROTATION_THRESHOLD_SECONDS: i64 = 8 * 60 * 60; + +pub struct ConnectConfig<'a> { + pub endpoint: &'a str, + pub root_ca_pem: &'a [u8], + pub timeout: Duration, +} + +pub struct ConnectClient { + endpoint: Url, + roots: RootCertStore, + root_certificates: Vec>, + client: Client, + timeout: Duration, +} + +impl ConnectClient { + pub fn from_optional_config(config: Option>) -> Result, ClientError> { + config.map(Self::new).transpose() + } + + pub fn new(config: ConnectConfig<'_>) -> Result { + let mut endpoint = Url::parse(config.endpoint).map_err(|_| ClientError::Endpoint)?; + if endpoint.scheme() != "https" + || endpoint.cannot_be_a_base() + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + { + return Err(ClientError::Endpoint); + } + if !endpoint.path().ends_with('/') { + let path = format!("{}/", endpoint.path()); + endpoint.set_path(&path); + } + + let root_certificates = CertificateDer::pem_slice_iter(config.root_ca_pem) + .collect::, _>>() + .map_err(|_| ClientError::RootCertificate)?; + if root_certificates.is_empty() { + return Err(ClientError::RootCertificate); + } + let mut roots = RootCertStore::empty(); + let (accepted, rejected) = roots.add_parsable_certificates(root_certificates.clone()); + if accepted != root_certificates.len() || rejected != 0 { + return Err(ClientError::RootCertificate); + } + + let client = build_client(&root_certificates, config.timeout, None)?; + Ok(Self { + endpoint, + roots, + root_certificates, + client, + timeout: config.timeout, + }) + } + + pub async fn register( + &self, + identity_store: &IdentityStore, + credential_store: &CredentialStore, + token: &RegistrationToken, + ) -> Result { + let _lock = credential_store.lock().await?; + if let Some((credential, _)) = self.load_valid_credential(identity_store, credential_store)? { + ensure_credential_time(&credential, unix_now())?; + return Ok(credential); + } + + let identity = identity_store.load_or_create()?; + let candidate = PendingRegistration { + token_uid: token.registration_token_uid.clone(), + request_id: Uuid::new_v4().to_string(), + certificate_request: identity.certificate_request_base64()?, + previous_credential_fingerprint: None, + next_public_key_sha256: None, + }; + let pending = credential_store.claim_pending_registration(&candidate)?; + if pending.token_uid != token.registration_token_uid + || pending.previous_credential_fingerprint.is_some() + || pending.next_public_key_sha256.is_some() + || !is_request_id(&pending.request_id) + || !certificate_request_matches(&pending.certificate_request, &identity)? + { + return Err(ClientError::PendingRegistration); + } + + let credential = match self.exchange_registration(token, &pending, &identity).await { + Ok(credential) => credential, + Err(error @ (ClientError::AccessRevoked { .. } | ClientError::Rejected { .. })) => { + credential_store.clear_pending_registration()?; + return Err(error); + } + Err(error) => return Err(error), + }; + credential_store.save(&credential)?; + credential_store.clear_pending_registration()?; + Ok(credential) + } + + pub async fn reenroll( + &self, + identity_store: &IdentityStore, + credential_store: &CredentialStore, + token: &RegistrationToken, + ) -> Result { + let _lock = credential_store.lock().await?; + let (credential, _) = self + .load_valid_credential(identity_store, credential_store)? + .ok_or(ClientError::NotRegistered)?; + let fingerprint = certificate_fingerprint(&credential.certificate)?; + if credential_store.load_completed_registration()?.is_some_and(|completed| { + completed.token_uid == token.registration_token_uid && completed.credential_fingerprint == fingerprint + }) { + return Ok(credential); + } + credential_store.clear_pending_rotation()?; + let next = identity_store.load_or_create_next()?; + let next_fingerprint = public_key_fingerprint(&next); + let candidate = PendingRegistration { + token_uid: token.registration_token_uid.clone(), + request_id: Uuid::new_v4().to_string(), + certificate_request: next.certificate_request_base64()?, + previous_credential_fingerprint: Some(fingerprint.clone()), + next_public_key_sha256: Some(next_fingerprint.clone()), + }; + let pending = credential_store.claim_pending_registration(&candidate)?; + if pending.token_uid != token.registration_token_uid + || pending.previous_credential_fingerprint.as_deref() != Some(&fingerprint) + || pending.next_public_key_sha256.as_deref() != Some(&next_fingerprint) + || !is_request_id(&pending.request_id) + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRegistration); + } + let enrolled = match self.exchange_registration(token, &pending, &next).await { + Ok(credential) => credential, + Err(error @ (ClientError::AccessRevoked { .. } | ClientError::Rejected { .. })) => { + credential_store.clear_pending_registration()?; + identity_store.clear_next()?; + return Err(error); + } + Err(error) => return Err(error), + }; + credential_store.save(&enrolled)?; + identity_store.commit_next(&next)?; + credential_store.save_completed_registration(&CompletedRegistration { + token_uid: token.registration_token_uid.clone(), + credential_fingerprint: certificate_fingerprint(&enrolled.certificate)?, + })?; + credential_store.clear_pending_registration()?; + Ok(enrolled) + } + + async fn exchange_registration( + &self, + token: &RegistrationToken, + pending: &PendingRegistration, + identity: &super::identity::DeviceIdentity, + ) -> Result { + let csr_der = base64::engine::general_purpose::STANDARD + .decode(&pending.certificate_request) + .map_err(|_| ClientError::PendingRegistration)?; + let transcript = RegistrationTranscript::build( + &token.registration_token_uid, + &token.organization_uid, + &token.cluster_uid, + &pending.request_id, + &token.challenge_nonce, + token.expires_unix, + &csr_der, + )?; + let proof = identity.sign_registration(&transcript); + let body = RegistrationRequest::new(token, &pending.request_id, &pending.certificate_request, &proof); + let url = self.url("./registrationTokens:exchange")?; + let response = self + .send(StatusCode::CREATED, || self.client.post(url.clone()).json(&body)) + .await?; + let cluster = format!("organizations/{}/clusters/{}", token.organization_uid, token.cluster_uid); + let credential = validate_credential( + response, + identity, + &self.roots, + &self.root_certificates, + ExpectedDevice::Registration { cluster: &cluster }, + )?; + Ok(credential) + } + + pub async fn rotate_if_due( + &self, + identity_store: &IdentityStore, + credential_store: &CredentialStore, + now_unix: i64, + ) -> Result, ClientError> { + let _lock = credential_store.lock().await?; + let (credential, identity) = self + .load_valid_credential(identity_store, credential_store)? + .ok_or(ClientError::NotRegistered)?; + if credential_store.load_pending_registration()?.is_some() { + return Err(ClientError::PendingRegistration); + } + ensure_credential_time(&credential, now_unix)?; + if credential.not_after_unix - now_unix > ROTATION_THRESHOLD_SECONDS { + return Ok(None); + } + let fingerprint = certificate_fingerprint(&credential.certificate)?; + let next = identity_store.load_or_create_next()?; + let candidate = PendingRotation { + credential_fingerprint: fingerprint.clone(), + device_name: credential.name.clone(), + request_id: Uuid::new_v4().to_string(), + certificate_request: next.certificate_request_base64()?, + next_public_key_sha256: public_key_fingerprint(&next), + }; + let pending = credential_store.claim_pending_rotation(&candidate)?; + if pending.credential_fingerprint != fingerprint + || pending.device_name != credential.name + || !is_request_id(&pending.request_id) + || pending.next_public_key_sha256 != public_key_fingerprint(&next) + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRotation); + } + let body = RotationRequest::new( + &identity, + &fingerprint, + &credential.name, + &pending.request_id, + &pending.certificate_request, + )?; + let private_key = identity.to_pkcs8_pem()?; + let mut identity_pem = Zeroizing::new(Vec::with_capacity(credential.certificate_chain.len() + private_key.len() + 1)); + identity_pem.extend_from_slice(credential.certificate_chain.as_bytes()); + identity_pem.push(b'\n'); + identity_pem.extend_from_slice(private_key.as_bytes()); + let tls_identity = reqwest::Identity::from_pem(&identity_pem).map_err(|_| ClientError::IdentityCertificate)?; + let client = build_client(&self.root_certificates, self.timeout, Some(tls_identity))?; + let path = format!("clusterDevices/{}:rotateCredential", credential.uid); + let url = self.url(&path)?; + let response = self.send(StatusCode::OK, || client.post(url.clone()).json(&body)).await?; + let rotated = validate_credential( + response, + &next, + &self.roots, + &self.root_certificates, + ExpectedDevice::Rotation { name: &credential.name }, + )?; + if rotated.name != credential.name || rotated.uid != credential.uid { + return Err(ClientError::Credential(CredentialValidationError::Identity)); + } + credential_store.save(&rotated)?; + identity_store.commit_next(&next)?; + credential_store.clear_pending_rotation()?; + Ok(Some(rotated)) + } + + fn load_valid_credential( + &self, + identity_store: &IdentityStore, + credential_store: &CredentialStore, + ) -> Result, ClientError> { + let Some(credential) = credential_store.load()? else { + return Ok(None); + }; + let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; + if let Some(pending) = credential_store.load_pending_registration()? { + let Some(previous) = pending.previous_credential_fingerprint.as_deref() else { + if pending.next_public_key_sha256.is_some() + || !is_request_id(&pending.request_id) + || !certificate_request_matches(&pending.certificate_request, ¤t)? + { + return Err(ClientError::PendingRegistration); + } + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + credential_store.clear_pending_registration()?; + return Ok(Some((credential, current))); + }; + let next_fingerprint = pending + .next_public_key_sha256 + .as_deref() + .ok_or(ClientError::PendingRegistration)?; + if !is_request_id(&pending.request_id) { + return Err(ClientError::PendingRegistration); + } + let fingerprint = certificate_fingerprint(&credential.certificate)?; + if fingerprint == previous { + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?; + if public_key_fingerprint(&next) != next_fingerprint + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRegistration); + } + return Ok(Some((credential, current))); + } + if public_key_fingerprint(¤t) == next_fingerprint { + if !certificate_request_matches(&pending.certificate_request, ¤t)? { + return Err(ClientError::PendingRegistration); + } + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + } else { + let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?; + if public_key_fingerprint(&next) != next_fingerprint + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRegistration); + } + validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?; + identity_store.commit_next(&next)?; + } + credential_store.save_completed_registration(&CompletedRegistration { + token_uid: pending.token_uid, + credential_fingerprint: fingerprint, + })?; + credential_store.clear_pending_registration()?; + let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; + return Ok(Some((credential, current))); + } + let Some(pending) = credential_store.load_pending_rotation()? else { + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + return Ok(Some((credential, current))); + }; + let fingerprint = certificate_fingerprint(&credential.certificate)?; + if pending.device_name != credential.name || !is_request_id(&pending.request_id) { + return Err(ClientError::PendingRotation); + } + if fingerprint == pending.credential_fingerprint { + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?; + if pending.next_public_key_sha256 != public_key_fingerprint(&next) + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRotation); + } + return Ok(Some((credential, current))); + } + + if public_key_fingerprint(¤t) == pending.next_public_key_sha256 { + if !certificate_request_matches(&pending.certificate_request, ¤t)? { + return Err(ClientError::PendingRotation); + } + validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + } else { + let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?; + if public_key_fingerprint(&next) != pending.next_public_key_sha256 + || !certificate_request_matches(&pending.certificate_request, &next)? + { + return Err(ClientError::PendingRotation); + } + validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?; + identity_store.commit_next(&next)?; + } + credential_store.clear_pending_rotation()?; + let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; + Ok(Some((credential, current))) + } + + async fn send(&self, success: StatusCode, mut request: F) -> Result + where + F: FnMut() -> reqwest::RequestBuilder, + { + let mut last_status = None; + for attempt in 0..MAX_ATTEMPTS { + match request().send().await { + Ok(response) if response.status() == success => return decode_response(response).await, + Ok(response) if matches!(response.status(), StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) => { + let status = response.status(); + let reason = decode_reason(response).await; + return Err(ClientError::AccessRevoked { status, reason }); + } + Ok(response) if matches!(response.status(), StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS) => { + last_status = Some(response.status()); + } + Ok(response) if response.status().is_client_error() => { + let status = response.status(); + let reason = decode_reason(response).await; + return Err(ClientError::Rejected { status, reason }); + } + Ok(response) if response.status().is_server_error() => { + last_status = Some(response.status()); + } + Ok(response) => { + let status = response.status(); + let reason = decode_reason(response).await; + return Err(ClientError::Rejected { status, reason }); + } + Err(error) if !error.is_timeout() && !error.is_connect() => return Err(ClientError::Transport(error)), + Err(_) => {} + } + + if attempt + 1 < MAX_ATTEMPTS { + tokio::time::sleep(Duration::from_millis(50 * (attempt as u64 + 1))).await; + } + } + Err(ClientError::Unavailable { status: last_status }) + } + + fn url(&self, path: &str) -> Result { + self.endpoint.join(path).map_err(|_| ClientError::Endpoint) + } +} + +fn is_request_id(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(uuid::Version::Random) && uuid.to_string() == value) +} + +fn unix_now() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs() as i64) +} + +fn ensure_credential_time(credential: &DeviceCredential, now_unix: i64) -> Result<(), ClientError> { + if credential.not_before_unix > now_unix { + return Err(ClientError::CredentialNotYetValid); + } + if credential.not_after_unix <= now_unix { + return Err(ClientError::CredentialExpired); + } + Ok(()) +} + +fn build_client( + roots: &[CertificateDer<'static>], + timeout: Duration, + identity: Option, +) -> Result { + let certificates = roots + .iter() + .map(|root| reqwest::Certificate::from_der(root.as_ref())) + .collect::, _>>()?; + let mut builder = Client::builder() + .https_only(true) + .redirect(reqwest::redirect::Policy::none()) + .timeout(timeout) + .tls_certs_only(certificates); + if let Some(identity) = identity { + builder = builder.identity(identity); + } + builder.build().map_err(ClientError::Transport) +} + +async fn decode_response(mut response: reqwest::Response) -> Result { + let body = read_body(&mut response).await?; + serde_json::from_slice(&body).map_err(|_| ClientError::Response) +} + +async fn read_body(response: &mut reqwest::Response) -> Result, ClientError> { + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(ClientError::Transport)? { + if body.len() + chunk.len() > MAX_RESPONSE_BYTES { + return Err(ClientError::ResponseTooLarge); + } + body.extend_from_slice(&chunk); + } + Ok(body) +} + +#[derive(Deserialize)] +struct ErrorEnvelope { + #[serde(default)] + details: Vec, +} + +#[derive(Deserialize)] +struct ErrorDetail { + #[serde(default)] + reason: String, +} + +async fn decode_reason(mut response: reqwest::Response) -> Option { + let body = read_body(&mut response).await.ok()?; + serde_json::from_slice::(&body) + .ok()? + .details + .into_iter() + .find_map(|detail| (!detail.reason.is_empty()).then_some(detail.reason)) +} + +#[derive(Debug, thiserror::Error)] +pub enum ClientError { + #[error("Connect endpoint must be an HTTPS base URL without credentials, query, or fragment")] + Endpoint, + #[error("Connect root CA configuration is invalid")] + RootCertificate, + #[error( + "Connect registration has a pending attempt for a different token; restore the original protected token configuration" + )] + PendingRegistration, + #[error( + "Connect credential rotation has an unfinished attempt for a different current certificate; inspect the local credential store" + )] + PendingRotation, + #[error("RustFS is not registered with Connect")] + NotRegistered, + #[error("the Connect device private key is missing; restore device.key before using the stored certificate")] + IdentityMissing, + #[error("the Connect device certificate has expired; call ConnectClient::reenroll with a fresh registration token")] + CredentialExpired, + #[error("the Connect device certificate is not yet valid; fix local clock skew or call ConnectClient::reenroll")] + CredentialNotYetValid, + #[error("the stored Connect certificate and device private key cannot form a TLS identity")] + IdentityCertificate, + #[error( + "Connect rejected the device credential with HTTP {status}; reason={reason:?}; call ConnectClient::reenroll with a fresh registration token if revoked" + )] + AccessRevoked { status: StatusCode, reason: Option }, + #[error("Connect rejected the request with HTTP {status}; reason={reason:?}")] + Rejected { status: StatusCode, reason: Option }, + #[error("Connect remained unavailable after bounded retries; last_status={status:?}")] + Unavailable { status: Option }, + #[error("Connect response exceeded the 1 MiB credential-response limit")] + ResponseTooLarge, + #[error("Connect returned an invalid credential response")] + Response, + #[error(transparent)] + Transport(#[from] reqwest::Error), + #[error(transparent)] + Identity(#[from] IdentityError), + #[error(transparent)] + IdentityStore(#[from] StoreError), + #[error(transparent)] + CredentialStore(#[from] CredentialStoreError), + #[error(transparent)] + Credential(#[from] CredentialValidationError), +} diff --git a/rustfs/src/connect/credential_store.rs b/rustfs/src/connect/credential_store.rs new file mode 100644 index 000000000..863ae1096 --- /dev/null +++ b/rustfs/src/connect/credential_store.rs @@ -0,0 +1,370 @@ +// 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. + +use std::fs; +use std::io::{self, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::{Deserialize, Serialize}; + +const CREDENTIAL_FILE: &str = "device.crt.json"; +const REGISTRATION_COMPLETED_FILE: &str = "registration.completed.json"; +const REGISTRATION_PENDING_FILE: &str = "registration.pending.json"; +const ROTATION_PENDING_FILE: &str = "rotation.pending.json"; +const LOCK_FILE: &str = ".state.lock"; + +#[cfg(unix)] +const FILE_MODE: u32 = 0o600; + +static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DeviceCredential { + pub name: String, + pub uid: String, + pub protocol_version: String, + pub key_id: String, + pub certificate_serial: String, + pub certificate: String, + pub certificate_chain: String, + pub not_before_unix: i64, + pub not_after_unix: i64, +} + +impl std::fmt::Debug for DeviceCredential { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DeviceCredential") + .field("name", &self.name) + .field("key_id", &self.key_id) + .field("certificate_serial", &self.certificate_serial) + .field("not_after_unix", &self.not_after_unix) + .finish_non_exhaustive() + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PendingRegistration { + pub token_uid: String, + pub request_id: String, + pub certificate_request: String, + #[serde(default)] + pub previous_credential_fingerprint: Option, + #[serde(default)] + pub next_public_key_sha256: Option, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CompletedRegistration { + pub token_uid: String, + pub credential_fingerprint: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PendingRotation { + pub credential_fingerprint: String, + pub device_name: String, + pub request_id: String, + pub certificate_request: String, + pub next_public_key_sha256: String, +} + +#[derive(Debug, thiserror::Error)] +pub enum CredentialStoreError { + #[error("connect credential store I/O failed at {path}: {source}")] + Io { + path: PathBuf, + #[source] + source: io::Error, + }, + + #[error("connect credential data at {path} is invalid: {source}")] + Invalid { + path: PathBuf, + #[source] + source: serde_json::Error, + }, + + #[cfg(unix)] + #[error("connect credential file at {path} has mode {mode:o}, expected {expected:o}")] + Permissions { path: PathBuf, mode: u32, expected: u32 }, +} + +#[derive(Clone, Debug)] +pub struct CredentialStore { + directory: PathBuf, +} + +pub(crate) struct CredentialLock { + _file: fs::File, +} + +impl CredentialStore { + pub fn new(directory: impl Into) -> Self { + Self { + directory: directory.into(), + } + } + + pub(crate) fn load(&self) -> Result, CredentialStoreError> { + self.read(CREDENTIAL_FILE) + } + + pub(crate) async fn lock(&self) -> Result { + let directory = self.directory.clone(); + tokio::task::spawn_blocking(move || { + fs::create_dir_all(&directory).map_err(|source| CredentialStoreError::Io { + path: directory.clone(), + source, + })?; + let path = directory.join(LOCK_FILE); + let mut options = fs::OpenOptions::new(); + options.create(true).truncate(false).read(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(FILE_MODE); + } + let file = options.open(&path).map_err(|source| CredentialStoreError::Io { + path: path.clone(), + source, + })?; + check_mode(&path)?; + file.lock().map_err(|source| CredentialStoreError::Io { path, source })?; + Ok(CredentialLock { _file: file }) + }) + .await + .map_err(|source| CredentialStoreError::Io { + path: self.directory.join(LOCK_FILE), + source: io::Error::other(source), + })? + } + + pub(crate) fn save(&self, credential: &DeviceCredential) -> Result<(), CredentialStoreError> { + self.write(CREDENTIAL_FILE, credential) + } + + pub(crate) fn claim_pending_registration( + &self, + pending: &PendingRegistration, + ) -> Result { + self.claim(REGISTRATION_PENDING_FILE, pending) + } + + pub(crate) fn load_pending_registration(&self) -> Result, CredentialStoreError> { + self.read(REGISTRATION_PENDING_FILE) + } + + pub(crate) fn clear_pending_registration(&self) -> Result<(), CredentialStoreError> { + self.remove(REGISTRATION_PENDING_FILE) + } + + pub(crate) fn load_completed_registration(&self) -> Result, CredentialStoreError> { + self.read(REGISTRATION_COMPLETED_FILE) + } + + pub(crate) fn save_completed_registration(&self, completed: &CompletedRegistration) -> Result<(), CredentialStoreError> { + self.write(REGISTRATION_COMPLETED_FILE, completed) + } + + pub(crate) fn load_pending_rotation(&self) -> Result, CredentialStoreError> { + self.read(ROTATION_PENDING_FILE) + } + + pub(crate) fn claim_pending_rotation(&self, pending: &PendingRotation) -> Result { + self.claim(ROTATION_PENDING_FILE, pending) + } + + pub(crate) fn clear_pending_rotation(&self) -> Result<(), CredentialStoreError> { + self.remove(ROTATION_PENDING_FILE) + } + + fn read Deserialize<'de>>(&self, file: &str) -> Result, CredentialStoreError> { + let path = self.directory.join(file); + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(source) => return Err(CredentialStoreError::Io { path, source }), + }; + + check_mode(&path)?; + serde_json::from_slice(&bytes) + .map(Some) + .map_err(|source| CredentialStoreError::Invalid { path, source }) + } + + fn write(&self, file: &str, value: &T) -> Result<(), CredentialStoreError> { + let bytes = serde_json::to_vec(value).map_err(|source| CredentialStoreError::Invalid { + path: self.directory.join(file), + source, + })?; + + fs::create_dir_all(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + })?; + + let final_path = self.directory.join(file); + let temp_path = self.stage(file, &bytes)?; + let result = fs::rename(&temp_path, &final_path) + .map_err(|source| CredentialStoreError::Io { + path: final_path, + source, + }) + .and_then(|()| { + fsync_dir(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + }) + }); + + if result.is_err() { + let _ = fs::remove_file(&temp_path); + } + result + } + + fn claim(&self, file: &str, value: &T) -> Result + where + T: Clone + Serialize + for<'de> Deserialize<'de>, + { + if let Some(existing) = self.read(file)? { + return Ok(existing); + } + let bytes = serde_json::to_vec(value).map_err(|source| CredentialStoreError::Invalid { + path: self.directory.join(file), + source, + })?; + fs::create_dir_all(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + })?; + let final_path = self.directory.join(file); + let temp_path = self.stage(file, &bytes)?; + let published = fs::hard_link(&temp_path, &final_path); + let _ = fs::remove_file(&temp_path); + match published { + Ok(()) => { + fsync_dir(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + })?; + Ok(value.clone()) + } + Err(source) if source.kind() == io::ErrorKind::AlreadyExists => { + fsync_dir(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + })?; + self.read(file)?.ok_or_else(|| CredentialStoreError::Io { + path: final_path, + source: io::Error::new(io::ErrorKind::NotFound, "pending state vanished after publication"), + }) + } + Err(source) => Err(CredentialStoreError::Io { + path: final_path, + source, + }), + } + } + + fn stage(&self, file: &str, bytes: &[u8]) -> Result { + loop { + let path = self.directory.join(format!( + ".{file}.{}.{}.tmp", + std::process::id(), + STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed) + )); + let mut options = fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(FILE_MODE); + } + let mut staging = match options.open(&path) { + Ok(staging) => staging, + Err(source) if source.kind() == io::ErrorKind::AlreadyExists => continue, + Err(source) => return Err(CredentialStoreError::Io { path, source }), + }; + let result = staging + .write_all(bytes) + .and_then(|()| staging.sync_all()) + .map_err(|source| CredentialStoreError::Io { + path: path.clone(), + source, + }) + .and_then(|()| check_mode(&path)); + if let Err(error) = result { + let _ = fs::remove_file(&path); + return Err(error); + } + return Ok(path); + } + } + + fn remove(&self, file: &str) -> Result<(), CredentialStoreError> { + let path = self.directory.join(file); + match fs::remove_file(&path) { + Ok(()) => fsync_dir(&self.directory).map_err(|source| CredentialStoreError::Io { + path: self.directory.clone(), + source, + }), + Err(source) if source.kind() == io::ErrorKind::NotFound => Ok(()), + Err(source) => Err(CredentialStoreError::Io { path, source }), + } + } +} + +#[cfg(unix)] +fn check_mode(path: &Path) -> Result<(), CredentialStoreError> { + use std::os::unix::fs::PermissionsExt as _; + + let mode = fs::metadata(path) + .map_err(|source| CredentialStoreError::Io { + path: path.to_path_buf(), + source, + })? + .permissions() + .mode() + & 0o7777; + if mode != FILE_MODE { + return Err(CredentialStoreError::Permissions { + path: path.to_path_buf(), + mode, + expected: FILE_MODE, + }); + } + Ok(()) +} + +#[cfg(not(unix))] +fn check_mode(_path: &Path) -> Result<(), CredentialStoreError> { + Ok(()) +} + +fn fsync_dir(directory: &Path) -> io::Result<()> { + #[cfg(unix)] + fs::File::open(directory)?.sync_all()?; + #[cfg(not(unix))] + let _ = directory; + Ok(()) +} diff --git a/rustfs/src/connect/identity.rs b/rustfs/src/connect/identity.rs index 64aa79266..2220f4e8c 100644 --- a/rustfs/src/connect/identity.rs +++ b/rustfs/src/connect/identity.rs @@ -24,7 +24,7 @@ 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 p256::pkcs8::{DecodePrivateKey as _, EncodePrivateKey as _, LineEnding}; use sha2::{Digest as _, Sha256}; use zeroize::Zeroizing; @@ -213,6 +213,12 @@ impl DeviceIdentity { .map_err(|error| IdentityError::MalformedKey(error.to_string())) } + pub(crate) fn to_pkcs8_pem(&self) -> Result, IdentityError> { + self.signing_key + .to_pkcs8_pem(LineEnding::LF) + .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 diff --git a/rustfs/src/connect/identity_store.rs b/rustfs/src/connect/identity_store.rs index 2f4e6724e..426c98352 100644 --- a/rustfs/src/connect/identity_store.rs +++ b/rustfs/src/connect/identity_store.rs @@ -32,6 +32,7 @@ use super::identity::{DeviceIdentity, IdentityError}; /// Name of the key file inside the store directory. const KEY_FILE: &str = "device.key"; +const NEXT_KEY_FILE: &str = "device.key.next"; /// Owner read/write only. The key is the device's whole identity. #[cfg(unix)] @@ -94,7 +95,15 @@ impl IdentityStore { /// 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, StoreError> { - let path = self.key_path(); + self.load_file(KEY_FILE) + } + + pub(crate) fn load_next(&self) -> Result, StoreError> { + self.load_file(NEXT_KEY_FILE) + } + + fn load_file(&self, file: &str) -> Result, StoreError> { + let path = self.directory.join(file); let der = match fs::read(&path) { Ok(der) => Zeroizing::new(der), @@ -142,11 +151,15 @@ impl IdentityStore { let candidate = DeviceIdentity::generate(); let der = candidate.to_pkcs8_der()?; - match self.publish(&der) { + match self.publish(KEY_FILE, &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 => { + fsync_dir(&self.directory).map_err(|source| StoreError::Io { + path: self.directory.clone(), + source, + })?; self.load()?.ok_or_else(|| StoreError::Io { path: self.key_path(), source: io::Error::new( @@ -159,35 +172,106 @@ impl IdentityStore { } } + pub(crate) fn load_or_create_next(&self) -> Result { + if let Some(identity) = self.load_next()? { + 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(NEXT_KEY_FILE, &der) { + Ok(()) => Ok(candidate), + Err(StoreError::Io { source, .. }) if source.kind() == io::ErrorKind::AlreadyExists => { + fsync_dir(&self.directory).map_err(|source| StoreError::Io { + path: self.directory.clone(), + source, + })?; + self.load_next()?.ok_or_else(|| StoreError::Io { + path: self.directory.join(NEXT_KEY_FILE), + source: io::Error::new(io::ErrorKind::NotFound, "next device key vanished after publication"), + }) + } + Err(error) => Err(error), + } + } + + pub(crate) fn commit_next(&self, expected: &DeviceIdentity) -> Result<(), StoreError> { + let next_path = self.directory.join(NEXT_KEY_FILE); + match fs::rename(&next_path, self.key_path()) { + Ok(()) => fsync_dir(&self.directory).map_err(|source| StoreError::Io { + path: self.directory.clone(), + source, + }), + Err(source) if source.kind() == io::ErrorKind::NotFound => { + let current = self.load()?.ok_or_else(|| StoreError::Io { + path: self.key_path(), + source, + })?; + if current.public_key_der() == expected.public_key_der() { + fsync_dir(&self.directory).map_err(|source| StoreError::Io { + path: self.directory.clone(), + source, + }) + } else { + Err(StoreError::Io { + path: next_path, + source: io::Error::new(io::ErrorKind::NotFound, "next device key is missing"), + }) + } + } + Err(source) => Err(StoreError::Io { path: next_path, source }), + } + } + + pub(crate) fn clear_next(&self) -> Result<(), StoreError> { + let path = self.directory.join(NEXT_KEY_FILE); + match fs::remove_file(&path) { + Ok(()) => fsync_dir(&self.directory).map_err(|source| StoreError::Io { + path: self.directory.clone(), + source, + }), + Err(source) if source.kind() == io::ErrorKind::NotFound => Ok(()), + Err(source) => Err(StoreError::Io { path, source }), + } + } + /// 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> { + fn publish(&self, file: &str, 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 final_path = self.directory.join(file); 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 (temp_path, mut staging) = loop { + let temp_path = self.directory.join(format!( + ".{file}.{}.{}.tmp", + std::process::id(), + STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed) + )); + let mut options = fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(KEY_MODE); + } + match options.open(&temp_path) { + Ok(staging) => break (temp_path, staging), + Err(source) if source.kind() == io::ErrorKind::AlreadyExists => continue, + Err(source) => return Err(io_at(&temp_path)(source)), + } + }; let result = (|| -> Result<(), StoreError> { - file.write_all(der).map_err(io_at(&temp_path))?; + staging.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. @@ -195,9 +279,10 @@ impl IdentityStore { { use std::os::unix::fs::PermissionsExt as _; - file.set_permissions(fs::Permissions::from_mode(KEY_MODE)) + staging + .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; + let mode = staging.metadata().map_err(io_at(&temp_path))?.permissions().mode() & 0o7777; if mode != KEY_MODE { return Err(StoreError::Permissions { path: temp_path.clone(), @@ -207,11 +292,11 @@ impl IdentityStore { } } - file.sync_all().map_err(io_at(&temp_path))?; + staging.sync_all().map_err(io_at(&temp_path))?; Ok(()) })(); - drop(file); + drop(staging); if let Err(error) = result { let _ = fs::remove_file(&temp_path); diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 8506055f2..350cceb38 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -25,10 +25,16 @@ //! 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 client; +pub mod credential_store; pub mod identity; pub mod identity_store; pub mod offline; +pub mod registration; +pub use client::{ClientError, ConnectClient, ConnectConfig}; +pub use credential_store::{CredentialStore, DeviceCredential}; pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript}; pub use identity_store::{IdentityStore, StoreError}; pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge}; +pub use registration::{RegistrationToken, TokenError}; diff --git a/rustfs/src/connect/registration.rs b/rustfs/src/connect/registration.rs new file mode 100644 index 000000000..0dbdbe122 --- /dev/null +++ b/rustfs/src/connect/registration.rs @@ -0,0 +1,527 @@ +// 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. + +use std::io::Read; +use std::sync::Arc; + +use base64::Engine as _; +use base64::engine::general_purpose::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 _; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, UnixTime, pem::PemObject as _}; +use rustls::server::WebPkiClientVerifier; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; +use uuid::{Uuid, Version}; +use x509_parser::extensions::GeneralName; +use x509_parser::oid_registry::OID_SIG_ECDSA_WITH_SHA256; +use x509_parser::prelude::{FromDer as _, X509Certificate, X509CertificationRequest}; +use zeroize::Zeroizing; + +use super::credential_store::DeviceCredential; +use super::identity::{DeviceIdentity, RegistrationProof}; + +pub const PROTOCOL_VERSION: &str = "v1"; + +const CERTIFICATE_LIFETIME_SECONDS: i64 = 86_400; +const ROTATION_DOMAIN: &[u8] = b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1"; +const MAX_TOKEN_BYTES: u64 = 16 * 1024; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct RegistrationTokenDocument { + registration_token_uid: String, + registration_token_secret: String, + organization_uid: String, + cluster_uid: String, + challenge_nonce: String, + expires_unix: i64, +} + +pub struct RegistrationToken { + pub registration_token_uid: String, + registration_token_secret: Zeroizing, + pub organization_uid: String, + pub cluster_uid: String, + pub challenge_nonce: String, + pub expires_unix: i64, +} + +impl std::fmt::Debug for RegistrationToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RegistrationToken") + .field("registration_token_uid", &self.registration_token_uid) + .field("expires_unix", &self.expires_unix) + .finish_non_exhaustive() + } +} + +impl RegistrationToken { + pub fn from_reader(reader: impl Read) -> Result { + let mut bytes = Zeroizing::new(Vec::new()); + reader + .take(MAX_TOKEN_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(TokenError::Read)?; + if bytes.len() as u64 > MAX_TOKEN_BYTES { + return Err(TokenError::TooLarge); + } + let document: RegistrationTokenDocument = serde_json::from_slice(&bytes).map_err(TokenError::Invalid)?; + let decoded = BASE64_URL_NO_PAD + .decode(&document.registration_token_secret) + .map(Zeroizing::new) + .map_err(|_| TokenError::SecretShape)?; + if decoded.len() != 32 || BASE64_URL_NO_PAD.encode(&decoded) != document.registration_token_secret { + return Err(TokenError::SecretShape); + } + if !is_uuid_v7(&document.registration_token_uid) + || !is_uuid_v7(&document.organization_uid) + || !is_uuid_v7(&document.cluster_uid) + || document.expires_unix < 0 + || document.challenge_nonce.len() != 64 + || !document + .challenge_nonce + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(TokenError::Shape); + } + + Ok(Self { + registration_token_uid: document.registration_token_uid, + registration_token_secret: Zeroizing::new(document.registration_token_secret), + organization_uid: document.organization_uid, + cluster_uid: document.cluster_uid, + challenge_nonce: document.challenge_nonce, + expires_unix: document.expires_unix, + }) + } + + pub(crate) fn secret(&self) -> &str { + &self.registration_token_secret + } +} + +#[derive(Debug, thiserror::Error)] +pub enum TokenError { + #[error("failed to read the Connect registration token")] + Read(#[source] std::io::Error), + #[error("Connect registration token configuration is invalid")] + Invalid(#[source] serde_json::Error), + #[error("Connect registration token secret must be 32-byte unpadded base64url")] + SecretShape, + #[error("Connect registration token configuration exceeds 16 KiB")] + TooLarge, + #[error("Connect registration token fields do not match the protocol schema")] + Shape, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RegistrationRequest<'a> { + protocol_version: &'static str, + request_id: &'a str, + registration_token_uid: &'a str, + registration_token_secret: &'a str, + certificate_request: &'a str, + proof: ProofRef<'a>, +} + +#[derive(Serialize)] +struct ProofRef<'a> { + algorithm: &'a str, + value: &'a str, +} + +impl<'a> RegistrationRequest<'a> { + pub(crate) fn new( + token: &'a RegistrationToken, + request_id: &'a str, + certificate_request: &'a str, + proof: &'a RegistrationProof, + ) -> Self { + Self { + protocol_version: PROTOCOL_VERSION, + request_id, + registration_token_uid: &token.registration_token_uid, + registration_token_secret: token.secret(), + certificate_request, + proof: ProofRef { + algorithm: &proof.algorithm, + value: &proof.value, + }, + } + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RotationRequest<'a> { + protocol_version: &'static str, + request_id: &'a str, + certificate_request: &'a str, + proof: ProofOwned, +} + +#[derive(Serialize)] +struct ProofOwned { + algorithm: String, + value: String, +} + +impl<'a> RotationRequest<'a> { + pub(crate) fn new( + identity: &DeviceIdentity, + credential_fingerprint: &str, + device_name: &str, + request_id: &'a str, + certificate_request: &'a str, + ) -> Result { + let csr_der = base64::engine::general_purpose::STANDARD + .decode(certificate_request) + .map_err(|_| CredentialValidationError::CertificateRequest)?; + let csr_digest = BASE64_URL_NO_PAD.encode(Sha256::digest(&csr_der)); + let transcript = rotation_transcript(credential_fingerprint, device_name, request_id, &csr_digest)?; + let key = identity + .to_pkcs8_der() + .map_err(|_| CredentialValidationError::CertificateRequest)?; + let signing_key = SigningKey::from_pkcs8_der(&key).map_err(|_| CredentialValidationError::CertificateRequest)?; + let signature: Signature = signing_key.sign(&transcript); + let canonical = signature.normalize_s().unwrap_or(signature); + + Ok(Self { + protocol_version: PROTOCOL_VERSION, + request_id, + certificate_request, + proof: ProofOwned { + algorithm: "ES256".to_string(), + value: BASE64_URL_NO_PAD.encode(canonical.to_bytes()), + }, + }) + } +} + +fn rotation_transcript( + credential_fingerprint: &str, + device_name: &str, + request_id: &str, + csr_digest: &str, +) -> Result, CredentialValidationError> { + let fields = [credential_fingerprint, device_name, request_id, csr_digest]; + if fields + .iter() + .any(|field| !field.is_ascii() || field.as_bytes().contains(&b'\n')) + { + return Err(CredentialValidationError::RotationTranscript); + } + + let mut transcript = Vec::with_capacity(346); + transcript.extend_from_slice(ROTATION_DOMAIN); + transcript.push(b'\n'); + for field in fields { + transcript.extend_from_slice(field.len().to_string().as_bytes()); + transcript.push(b':'); + transcript.extend_from_slice(field.as_bytes()); + transcript.push(b'\n'); + } + Ok(transcript) +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct CredentialResponse { + pub name: String, + #[serde(default)] + pub uid: String, + #[serde(default)] + pub cluster: String, + pub protocol_version: String, + pub key_id: String, + pub certificate_serial: String, + pub certificate: String, + pub certificate_chain: String, + pub not_before: String, + pub not_after: String, +} + +pub(crate) enum ExpectedDevice<'a> { + Registration { cluster: &'a str }, + Rotation { name: &'a str }, + Stored, +} + +#[derive(Debug, thiserror::Error)] +pub enum CredentialValidationError { + #[error("Connect returned malformed certificate material")] + Certificate, + #[error("Connect returned a certificate chain that is not trusted")] + Chain, + #[error("Connect returned a certificate for the wrong device identity")] + Identity, + #[error("Connect returned a certificate for a different device key")] + Key, + #[error("Connect returned an invalid certificate validity window")] + Validity, + #[error("the device certificate request could not be prepared")] + CertificateRequest, + #[error("the credential rotation transcript contains an invalid field")] + RotationTranscript, +} + +pub(crate) fn validate_credential( + response: CredentialResponse, + identity: &DeviceIdentity, + roots: &RootCertStore, + root_certificates: &[CertificateDer<'static>], + expected: ExpectedDevice<'_>, +) -> Result { + validate_credential_at(response, identity, roots, root_certificates, expected, true) +} + +pub(crate) fn validate_stored_credential( + credential: &DeviceCredential, + identity: &DeviceIdentity, + roots: &RootCertStore, + root_certificates: &[CertificateDer<'static>], +) -> Result<(), CredentialValidationError> { + let not_before = OffsetDateTime::from_unix_timestamp(credential.not_before_unix) + .map_err(|_| CredentialValidationError::Validity)? + .format(&Rfc3339) + .map_err(|_| CredentialValidationError::Validity)?; + let not_after = OffsetDateTime::from_unix_timestamp(credential.not_after_unix) + .map_err(|_| CredentialValidationError::Validity)? + .format(&Rfc3339) + .map_err(|_| CredentialValidationError::Validity)?; + let response = CredentialResponse { + name: credential.name.clone(), + uid: credential.uid.clone(), + cluster: String::new(), + protocol_version: credential.protocol_version.clone(), + key_id: credential.key_id.clone(), + certificate_serial: credential.certificate_serial.clone(), + certificate: credential.certificate.clone(), + certificate_chain: credential.certificate_chain.clone(), + not_before, + not_after, + }; + validate_credential_at(response, identity, roots, root_certificates, ExpectedDevice::Stored, false).map(|_| ()) +} + +fn validate_credential_at( + response: CredentialResponse, + identity: &DeviceIdentity, + roots: &RootCertStore, + root_certificates: &[CertificateDer<'static>], + expected: ExpectedDevice<'_>, + verify_now: bool, +) -> Result { + if response.protocol_version != PROTOCOL_VERSION { + return Err(CredentialValidationError::Identity); + } + + let leaves = CertificateDer::pem_slice_iter(response.certificate.as_bytes()) + .collect::, _>>() + .map_err(|_| CredentialValidationError::Certificate)?; + if leaves.len() != 1 { + return Err(CredentialValidationError::Certificate); + } + let chain = CertificateDer::pem_slice_iter(response.certificate_chain.as_bytes()) + .collect::, _>>() + .map_err(|_| CredentialValidationError::Certificate)?; + if chain.is_empty() + || chain[0].as_ref() != leaves[0].as_ref() + || chain + .iter() + .skip(1) + .any(|certificate| root_certificates.iter().any(|root| root.as_ref() == certificate.as_ref())) + { + return Err(CredentialValidationError::Chain); + } + + let (remaining, certificate) = + X509Certificate::from_der(leaves[0].as_ref()).map_err(|_| CredentialValidationError::Certificate)?; + if !remaining.is_empty() { + return Err(CredentialValidationError::Certificate); + } + + let uid = match expected { + ExpectedDevice::Registration { cluster } => { + if response.uid.is_empty() + || response.cluster != cluster + || response.name != format!("{cluster}/clusterDevices/{}", response.uid) + { + return Err(CredentialValidationError::Identity); + } + response.uid.clone() + } + ExpectedDevice::Rotation { name } => { + if response.name != name || !response.uid.is_empty() || !response.cluster.is_empty() { + return Err(CredentialValidationError::Identity); + } + name.rsplit_once("/clusterDevices/") + .map(|(_, uid)| uid.to_string()) + .ok_or(CredentialValidationError::Identity)? + } + ExpectedDevice::Stored => { + let (cluster, name_uid) = response + .name + .rsplit_once("/clusterDevices/") + .ok_or(CredentialValidationError::Identity)?; + if response.uid != name_uid + || !response.cluster.is_empty() + || !valid_cluster_name(cluster) + || response.name.matches("/clusterDevices/").count() != 1 + { + return Err(CredentialValidationError::Identity); + } + response.uid.clone() + } + }; + let parsed_uid = Uuid::parse_str(&uid).map_err(|_| CredentialValidationError::Identity)?; + if parsed_uid.get_version() != Some(Version::SortRand) || parsed_uid.to_string() != uid { + return Err(CredentialValidationError::Identity); + } + let expected_uri = format!("urn:rustfs:connect:device:{uid}"); + let common_names = certificate + .subject() + .iter_common_name() + .map(|name| name.as_str()) + .collect::, _>>() + .map_err(|_| CredentialValidationError::Identity)?; + let san = certificate + .subject_alternative_name() + .map_err(|_| CredentialValidationError::Identity)? + .ok_or(CredentialValidationError::Identity)?; + let san_matches = matches!(san.value.general_names.as_slice(), [GeneralName::URI(uri)] if *uri == expected_uri); + + if common_names.as_slice() != [uid.as_str()] + || !san_matches + || certificate.subject().iter().count() != 1 + || certificate.subject().iter_attributes().count() != 1 + { + return Err(CredentialValidationError::Identity); + } + if certificate.public_key().raw != identity.public_key_der() { + return Err(CredentialValidationError::Key); + } + + let not_before = OffsetDateTime::parse(&response.not_before, &Rfc3339) + .map_err(|_| CredentialValidationError::Validity)? + .unix_timestamp(); + let not_after = OffsetDateTime::parse(&response.not_after, &Rfc3339) + .map_err(|_| CredentialValidationError::Validity)? + .unix_timestamp(); + let verify_unix = if verify_now { + UnixTime::now() + } else { + let midpoint = certificate.validity().not_before.timestamp() + + (certificate.validity().not_after.timestamp() - certificate.validity().not_before.timestamp()) / 2; + UnixTime::since_unix_epoch(std::time::Duration::from_secs( + midpoint.try_into().map_err(|_| CredentialValidationError::Validity)?, + )) + }; + let verifier = WebPkiClientVerifier::builder(Arc::new(roots.clone())) + .build() + .map_err(|_| CredentialValidationError::Chain)?; + verifier + .verify_client_cert(&leaves[0], &chain[1..], verify_unix) + .map_err(|_| CredentialValidationError::Chain)?; + if not_before != certificate.validity().not_before.timestamp() + || not_after != certificate.validity().not_after.timestamp() + || not_after - not_before != CERTIFICATE_LIFETIME_SECONDS + || certificate.signature_algorithm.algorithm != OID_SIG_ECDSA_WITH_SHA256 + || response.certificate_serial != canonical_serial(certificate.raw_serial())? + || response.key_id != format!("x509-{}", response.certificate_serial) + { + return Err(CredentialValidationError::Validity); + } + + Ok(DeviceCredential { + name: response.name, + uid, + protocol_version: response.protocol_version, + key_id: response.key_id, + certificate_serial: response.certificate_serial, + certificate: response.certificate, + certificate_chain: response.certificate_chain, + not_before_unix: not_before, + not_after_unix: not_after, + }) +} + +pub(crate) fn public_key_fingerprint(identity: &DeviceIdentity) -> String { + hex_lower(&Sha256::digest(identity.public_key_der())) +} + +pub(crate) fn certificate_request_matches(encoded: &str, identity: &DeviceIdentity) -> Result { + let der = base64::engine::general_purpose::STANDARD + .decode(encoded) + .map_err(|_| CredentialValidationError::CertificateRequest)?; + let (remaining, request) = + X509CertificationRequest::from_der(&der).map_err(|_| CredentialValidationError::CertificateRequest)?; + Ok(remaining.is_empty() && request.certification_request_info.subject_pki.raw == identity.public_key_der()) +} + +fn canonical_serial(raw: &[u8]) -> Result { + let magnitude = match raw { + [0, first, rest @ ..] if first & 0x80 != 0 => { + if rest.len() + 1 > 16 { + return Err(CredentialValidationError::Validity); + } + &raw[1..] + } + [0] => raw, + [0, ..] | [] => return Err(CredentialValidationError::Validity), + [first, ..] if first & 0x80 != 0 => return Err(CredentialValidationError::Validity), + _ if raw.len() > 16 => return Err(CredentialValidationError::Validity), + _ => raw, + }; + let mut padded = [0u8; 16]; + padded[16 - magnitude.len()..].copy_from_slice(magnitude); + Ok(hex_lower(&padded)) +} + +fn is_uuid_v7(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(Version::SortRand) && uuid.to_string() == value) +} + +fn valid_cluster_name(name: &str) -> bool { + let Some((organization, cluster)) = name + .strip_prefix("organizations/") + .and_then(|rest| rest.split_once("/clusters/")) + else { + return false; + }; + !cluster.contains('/') && is_uuid_v7(organization) && is_uuid_v7(cluster) +} + +pub(crate) fn certificate_fingerprint(certificate_pem: &str) -> Result { + let certificate = CertificateDer::pem_slice_iter(certificate_pem.as_bytes()) + .next() + .ok_or(CredentialValidationError::Certificate)? + .map_err(|_| CredentialValidationError::Certificate)?; + Ok(hex_lower(&Sha256::digest(certificate.as_ref()))) +} + +fn hex_lower(bytes: &[u8]) -> String { + bytes.iter().fold(String::with_capacity(bytes.len() * 2), |mut output, byte| { + use std::fmt::Write as _; + let _ = write!(output, "{byte:02x}"); + output + }) +} diff --git a/rustfs/tests/connect_registration.rs b/rustfs/tests/connect_registration.rs new file mode 100644 index 000000000..a1e94030d --- /dev/null +++ b/rustfs/tests/connect_registration.rs @@ -0,0 +1,911 @@ +// 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. + +use std::collections::VecDeque; +use std::fs; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use base64::Engine as _; +use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD}; +use bytes::Bytes; +use http_body_util::{BodyExt as _, Full}; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use hyper_util::rt::TokioIo; +use p256::ecdsa::signature::Verifier as _; +use p256::ecdsa::{Signature, VerifyingKey}; +use p256::pkcs8::DecodePublicKey as _; +use rcgen::{ + BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair, + KeyUsagePurpose, SanType, SerialNumber, +}; +use rustfs::connect::{ClientError, ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken, TokenError}; +use rustls::RootCertStore; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject as _}; +use rustls::server::WebPkiClientVerifier; +use serde_json::{Value, json}; +use sha2::{Digest as _, Sha256}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; +use tokio::net::TcpListener; +use tokio_rustls::TlsAcceptor; + +const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70"; +const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81"; +const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92"; +const TOKEN_UID: &str = "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5"; +const FRESH_TOKEN_UID: &str = "0198f4b0-7f00-7c70-a381-8e9fa0b1c2d6"; +const SECOND_TOKEN_UID: &str = "0198f4b0-8f00-7d80-b491-9fa0b1c2d3e7"; + +struct TestPki { + root_params: CertificateParams, + root_key: KeyPair, + root_der: CertificateDer<'static>, + root_pem: String, + server_der: CertificateDer<'static>, + server_key: PrivatePkcs8KeyDer<'static>, +} + +impl TestPki { + fn new() -> Self { + let now = OffsetDateTime::now_utc(); + let root_key = KeyPair::generate().expect("generate root key"); + let mut root_params = CertificateParams::default(); + root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + root_params.not_before = now - time::Duration::days(1); + root_params.not_after = now + time::Duration::days(30); + root_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::DigitalSignature]; + root_params.distinguished_name.push(DnType::CommonName, "Connect test root"); + let root = root_params.self_signed(&root_key).expect("sign root"); + + let server_key = KeyPair::generate().expect("generate server key"); + let mut server_params = CertificateParams::default(); + server_params.not_before = now - time::Duration::hours(1); + server_params.not_after = now + time::Duration::days(2); + server_params + .subject_alt_names + .push(SanType::DnsName("localhost".try_into().expect("valid DNS name"))); + server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth]; + let issuer = Issuer::from_params(&root_params, &root_key); + let server = server_params + .signed_by(&server_key, &issuer) + .expect("sign server certificate"); + + Self { + root_params, + root_key, + root_der: root.der().clone(), + root_pem: root.pem(), + server_der: server.der().clone(), + server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()), + } + } + + fn credential(&self, identity: &rustfs::connect::DeviceIdentity, uri: &str, serial_byte: u8) -> Value { + let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second"); + self.credential_window(identity, uri, serial_byte, now, now + time::Duration::days(1)) + } + + fn credential_window( + &self, + identity: &rustfs::connect::DeviceIdentity, + uri: &str, + serial_byte: u8, + not_before: OffsetDateTime, + not_after: OffsetDateTime, + ) -> Value { + let mut params = CertificateParams::default(); + params.not_before = not_before; + params.not_after = not_after; + params.serial_number = Some(SerialNumber::from(vec![serial_byte; 16])); + params.key_usages = vec![KeyUsagePurpose::DigitalSignature]; + params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth]; + params.distinguished_name = DistinguishedName::new(); + params.distinguished_name.push(DnType::CommonName, DEVICE_UID); + params + .subject_alt_names + .push(SanType::URI(uri.try_into().expect("valid URI SAN"))); + let private_key = identity.to_pkcs8_der().expect("serialize device key"); + let private_key = PrivatePkcs8KeyDer::from(private_key.to_vec()); + let device_key = + KeyPair::from_pkcs8_der_and_sign_algo(&private_key, &rcgen::PKCS_ECDSA_P256_SHA256).expect("parse device key"); + let issuer = Issuer::from_params(&self.root_params, &self.root_key); + let certificate = params.signed_by(&device_key, &issuer).expect("sign device certificate"); + let serial = format!("{serial_byte:02x}").repeat(16); + let cluster = format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}"); + + json!({ + "name": format!("{cluster}/clusterDevices/{DEVICE_UID}"), + "uid": DEVICE_UID, + "cluster": cluster, + "protocolVersion": "v1", + "keyId": format!("x509-{serial}"), + "certificateSerial": serial, + "certificate": certificate.pem(), + "certificateChain": certificate.pem(), + "notBefore": not_before.format(&Rfc3339).expect("format notBefore"), + "notAfter": not_after.format(&Rfc3339).expect("format notAfter"), + }) + } + + fn server_config(&self, require_client: bool) -> rustls::ServerConfig { + let mut roots = RootCertStore::empty(); + roots.add(self.root_der.clone()).expect("add client root"); + let verifier = WebPkiClientVerifier::builder(Arc::new(roots)); + let verifier = if require_client { + verifier.build() + } else { + verifier.allow_unauthenticated().build() + } + .expect("build client verifier"); + rustls::ServerConfig::builder() + .with_client_cert_verifier(verifier) + .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key())) + .expect("build server TLS") + } +} + +#[derive(Clone)] +enum Reply { + Json(StatusCode, Value), + DelayedClose(Duration), + VerifiedRotation { + response: Value, + current_public_key: Vec, + current_certificate_fingerprint: String, + device_name: String, + }, +} + +struct TestServer { + endpoint: String, + seen: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn server(pki: &TestPki, replies: Vec) -> TestServer { + server_with_client_auth(pki, replies, false).await +} + +async fn server_with_client_auth(pki: &TestPki, replies: Vec, require_client: bool) -> TestServer { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind test server"); + let address = listener.local_addr().expect("server address"); + let acceptor = TlsAcceptor::from(Arc::new(pki.server_config(require_client))); + let replies = Arc::new(Mutex::new(VecDeque::from(replies))); + let seen = Arc::new(Mutex::new(Vec::new())); + let captured = seen.clone(); + let task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let acceptor = acceptor.clone(); + let replies = replies.clone(); + let seen = captured.clone(); + tokio::spawn(async move { + let Ok(stream) = acceptor.accept(stream).await else { + return; + }; + let service = service_fn(move |request: Request| { + let replies = replies.clone(); + let seen = seen.clone(); + async move { + let body = request.into_body().collect().await.expect("read request body").to_bytes(); + let value: Value = serde_json::from_slice(&body).expect("request JSON"); + seen.lock().expect("seen lock").push(value.clone()); + let reply = replies.lock().expect("reply lock").pop_front().expect("planned reply"); + match reply { + Reply::Json(status, value) => Ok::<_, hyper::Error>( + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Full::new(Bytes::from(serde_json::to_vec(&value).expect("reply JSON")))) + .expect("response"), + ), + Reply::DelayedClose(delay) => { + tokio::time::sleep(delay).await; + Ok(Response::builder() + .status(StatusCode::SERVICE_UNAVAILABLE) + .body(Full::new(Bytes::new())) + .expect("response")) + } + Reply::VerifiedRotation { + response, + current_public_key, + current_certificate_fingerprint, + device_name, + } => { + verify_rotation_request( + &value, + ¤t_public_key, + ¤t_certificate_fingerprint, + &device_name, + ); + Ok(Response::builder() + .status(StatusCode::OK) + .header("content-type", "application/json") + .body(Full::new(Bytes::from(serde_json::to_vec(&response).expect("reply JSON")))) + .expect("response")) + } + } + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await; + }); + } + }); + TestServer { + endpoint: format!("https://localhost:{}/agent/", address.port()), + seen, + task, + } +} + +fn verify_rotation_request(request: &Value, current_public_key: &[u8], fingerprint: &str, device_name: &str) { + assert_eq!(request["protocolVersion"], "v1"); + assert_eq!(request["proof"]["algorithm"], "ES256"); + let csr = BASE64_STANDARD + .decode(request["certificateRequest"].as_str().expect("certificateRequest")) + .expect("CSR base64"); + let csr_digest = BASE64_URL_NO_PAD.encode(Sha256::digest(&csr)); + let request_id = request["requestId"].as_str().expect("requestId"); + let transcript = rebuilt_rotation_transcript( + b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1", + [fingerprint, device_name, request_id, &csr_digest], + ); + let encoded = request["proof"]["value"].as_str().expect("proof value"); + assert_eq!(encoded.len(), 86); + let raw = BASE64_URL_NO_PAD.decode(encoded).expect("proof base64url"); + let signature = Signature::from_slice(&raw).expect("fixed-width signature"); + assert!(signature.normalize_s().is_none(), "rotation proof must be low-S"); + let verifying = VerifyingKey::from_public_key_der(current_public_key).expect("current public key"); + verifying.verify(&transcript, &signature).expect("rotation proof verifies"); + + let wrong_domain = rebuilt_rotation_transcript( + b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V2", + [fingerprint, device_name, request_id, &csr_digest], + ); + assert!(verifying.verify(&wrong_domain, &signature).is_err()); + let wrong_order = rebuilt_rotation_transcript( + b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1", + [device_name, fingerprint, request_id, &csr_digest], + ); + assert!(verifying.verify(&wrong_order, &signature).is_err()); +} + +fn rebuilt_rotation_transcript(domain: &[u8], fields: [&str; 4]) -> Vec { + let mut transcript = Vec::new(); + transcript.extend_from_slice(domain); + transcript.push(b'\n'); + for field in fields { + transcript.extend_from_slice(field.len().to_string().as_bytes()); + transcript.push(b':'); + transcript.extend_from_slice(field.as_bytes()); + transcript.push(b'\n'); + } + transcript +} + +fn certificate_fingerprint(pem: &str) -> String { + let certificate = CertificateDer::pem_slice_iter(pem.as_bytes()) + .next() + .expect("leaf certificate") + .expect("certificate PEM"); + Sha256::digest(certificate.as_ref()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn token_document() -> Value { + json!({ + "registrationTokenUid": TOKEN_UID, + "registrationTokenSecret": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + "organizationUid": ORGANIZATION_UID, + "clusterUid": CLUSTER_UID, + "challengeNonce": "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0f", + "expiresUnix": OffsetDateTime::now_utc().unix_timestamp() + 3600, + }) +} + +fn token() -> RegistrationToken { + token_with_uid(TOKEN_UID) +} + +fn token_with_uid(uid: &str) -> RegistrationToken { + let document = token_document(); + let mut document = document; + document["registrationTokenUid"] = json!(uid); + RegistrationToken::from_reader(serde_json::to_vec(&document).expect("token JSON").as_slice()).expect("token parses") +} + +fn stores(temp: &tempfile::TempDir) -> (IdentityStore, CredentialStore) { + ( + IdentityStore::new(temp.path().join("identity")), + CredentialStore::new(temp.path().join("credential")), + ) +} + +fn client(server: &TestServer, pki: &TestPki, timeout: Duration) -> ConnectClient { + ConnectClient::new(ConnectConfig { + endpoint: &server.endpoint, + root_ca_pem: pki.root_pem.as_bytes(), + timeout, + }) + .expect("build Connect client") +} + +fn rotation_response(pki: &TestPki, identity: &rustfs::connect::DeviceIdentity, serial: u8) -> (Value, Value) { + let stored = pki.credential(identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), serial); + let mut wire = stored.clone(); + wire.as_object_mut().expect("response object").remove("uid"); + wire.as_object_mut().expect("response object").remove("cluster"); + (wire, stored) +} + +fn write_stored_credential(path: &std::path::Path, response: &Value) { + let not_before = OffsetDateTime::parse(response["notBefore"].as_str().expect("notBefore"), &Rfc3339) + .expect("parse notBefore") + .unix_timestamp(); + let not_after = OffsetDateTime::parse(response["notAfter"].as_str().expect("notAfter"), &Rfc3339) + .expect("parse notAfter") + .unix_timestamp(); + let stored = json!({ + "name": response["name"], + "uid": DEVICE_UID, + "protocolVersion": response["protocolVersion"], + "keyId": response["keyId"], + "certificateSerial": response["certificateSerial"], + "certificate": response["certificate"], + "certificateChain": response["certificateChain"], + "notBeforeUnix": not_before, + "notAfterUnix": not_after, + }); + fs::write(path, serde_json::to_vec(&stored).expect("stored credential JSON")).expect("write credential"); + set_owner_only(path); +} + +#[cfg(unix)] +fn set_owner_only(path: &std::path::Path) { + use std::os::unix::fs::PermissionsExt as _; + fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("set owner-only mode"); +} + +#[cfg(not(unix))] +fn set_owner_only(_path: &std::path::Path) {} + +#[tokio::test] +async fn registration_reuses_request_and_csr_after_timeout_and_restart() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let response = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x80); + let server = server( + &pki, + vec![ + Reply::DelayedClose(Duration::from_millis(200)), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::CREATED, response), + ], + ) + .await; + let first = client(&server, &pki, Duration::from_millis(80)); + assert!(matches!( + first.register(&identity_store, &credential_store, &token()).await, + Err(ClientError::Unavailable { .. }) + )); + + let restarted = client(&server, &pki, Duration::from_secs(2)); + let credential = restarted + .register(&identity_store, &credential_store, &token()) + .await + .expect("restart replays completed exchange"); + assert_eq!(credential.uid, DEVICE_UID); + assert_eq!(credential.certificate_serial, "80".repeat(16)); + + let seen = server.seen.lock().expect("seen lock"); + assert_eq!(seen.len(), 4); + for request in &seen[1..] { + assert_eq!(request["requestId"], seen[0]["requestId"]); + assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]); + } +} + +#[tokio::test] +async fn registration_rejects_untrusted_or_misbound_credentials() { + for case in ["san", "chain", "key", "key_id", "cluster", "name"] { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let mut response = match case { + "san" => pki.credential(&identity, "urn:rustfs:connect:device:0198f4b0-3c00-7e30-8f41-4a5b6c7d8e93", 2), + "chain" => TestPki::new().credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 2), + "key" => pki.credential( + &rustfs::connect::DeviceIdentity::generate(), + &format!("urn:rustfs:connect:device:{DEVICE_UID}"), + 2, + ), + _ => pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 2), + }; + match case { + "key_id" => response["keyId"] = json!("x509-deadbeef"), + "cluster" => response["cluster"] = json!(format!("organizations/{ORGANIZATION_UID}/clusters/other")), + "name" => { + response["name"] = json!(format!("organizations/{ORGANIZATION_UID}/clusters/other/clusterDevices/{DEVICE_UID}")) + } + _ => {} + } + let server = server(&pki, vec![Reply::Json(StatusCode::CREATED, response)]).await; + let error = client(&server, &pki, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect_err("invalid returned identity must fail closed"); + assert!(matches!(error, ClientError::Credential(_))); + assert!(!temp.path().join("credential/device.crt.json").exists()); + } +} + +#[tokio::test] +async fn stored_credential_is_revalidated_before_reuse() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 4); + let server = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + let client = client(&server, &pki, Duration::from_secs(2)); + client + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + + let path = temp.path().join("credential/device.crt.json"); + let mut stored: Value = serde_json::from_slice(&fs::read(&path).expect("read credential")).expect("credential JSON"); + stored["certificateSerial"] = json!("00".repeat(16)); + fs::write(&path, serde_json::to_vec(&stored).expect("credential JSON")).expect("tamper credential"); + let error = client + .register(&identity_store, &credential_store, &token()) + .await + .expect_err("tampered stored credential must fail closed"); + assert!(matches!(error, ClientError::Credential(_))); +} + +#[tokio::test] +async fn register_rejects_expired_and_not_yet_valid_stored_credentials() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 9); + let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + let client = client(®istration, &pki, Duration::from_secs(2)); + client + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + + let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second"); + let path = temp.path().join("credential/device.crt.json"); + let expired = pki.credential_window( + &identity, + &format!("urn:rustfs:connect:device:{DEVICE_UID}"), + 10, + now - time::Duration::days(2), + now - time::Duration::days(1), + ); + write_stored_credential(&path, &expired); + assert!(matches!( + client.register(&identity_store, &credential_store, &token()).await, + Err(ClientError::CredentialExpired) + )); + + let future = pki.credential_window( + &identity, + &format!("urn:rustfs:connect:device:{DEVICE_UID}"), + 11, + now + time::Duration::hours(1), + now + time::Duration::hours(25), + ); + write_stored_credential(&path, &future); + assert!(matches!( + client.register(&identity_store, &credential_store, &token()).await, + Err(ClientError::CredentialNotYetValid) + )); +} + +#[tokio::test] +async fn concurrent_rotation_retries_converge_and_promote_the_next_key() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let current = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let mut issued = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 5); + issued["certificateChain"] = json!( + issued["certificateChain"] + .as_str() + .expect("certificate chain") + .trim_end_matches('\n') + ); + let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + let registered = client(®istration, &pki, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + + let retries = server( + &pki, + vec![ + Reply::DelayedClose(Duration::from_millis(200)), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + ], + ) + .await; + let retry_client = client(&retries, &pki, Duration::from_millis(80)); + let due = registered.not_after_unix - 8 * 60 * 60; + let (first, second) = tokio::join!( + retry_client.rotate_if_due(&identity_store, &credential_store, due), + retry_client.rotate_if_due(&identity_store, &credential_store, due) + ); + assert!(matches!(first, Err(ClientError::Unavailable { .. }))); + assert!(matches!(second, Err(ClientError::Unavailable { .. }))); + let (request_id, certificate_request) = { + let seen = retries.seen.lock().expect("seen lock"); + assert!(seen.len() >= 3, "bounded retries must reach the server"); + for request in &seen[1..] { + assert_eq!(request["requestId"], seen[0]["requestId"]); + assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]); + } + (seen[0]["requestId"].clone(), seen[0]["certificateRequest"].clone()) + }; + + let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read staged next key"); + let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key"); + assert_ne!(current.public_key_der(), next.public_key_der()); + assert_eq!( + identity_store + .load() + .expect("load current key") + .expect("current key") + .public_key_der(), + current.public_key_der() + ); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + let mode = fs::metadata(temp.path().join("identity/device.key.next")) + .expect("next key metadata") + .permissions() + .mode() + & 0o7777; + assert_eq!(mode, 0o600); + } + let (rotated, _) = rotation_response(&pki, &next, 6); + let success = server_with_client_auth( + &pki, + vec![Reply::VerifiedRotation { + response: rotated, + current_public_key: current.public_key_der(), + current_certificate_fingerprint: certificate_fingerprint(®istered.certificate), + device_name: registered.name.clone(), + }], + true, + ) + .await; + let success_client = client(&success, &pki, Duration::from_secs(2)); + let (due_result, current_result) = tokio::join!( + success_client.rotate_if_due(&identity_store, &credential_store, due), + success_client.rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp()) + ); + let credential = due_result + .expect("retry rotation") + .or(current_result.expect("concurrent current-state check")) + .expect("exactly one rotation is due"); + assert_eq!(credential.certificate_serial, "06".repeat(16)); + let success_seen = success.seen.lock().expect("seen lock"); + assert_eq!(success_seen.len(), 1, "the post-commit actor must not publish stale state"); + assert_eq!(success_seen[0]["requestId"], request_id); + assert_eq!(success_seen[0]["certificateRequest"], certificate_request); + drop(success_seen); + assert_eq!( + identity_store + .load() + .expect("load key") + .expect("current key") + .public_key_der(), + next.public_key_der() + ); + assert!(!temp.path().join("identity/device.key.next").exists()); + assert!(!temp.path().join("credential/rotation.pending.json").exists()); +} + +#[tokio::test] +async fn rotation_commit_recovers_after_each_durable_step() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let current = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 7); + let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + let registered = client(®istration, &pki, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; + let due = registered.not_after_unix - 8 * 60 * 60; + assert!(matches!( + client(&failed, &pki, Duration::from_secs(2)) + .rotate_if_due(&identity_store, &credential_store, due) + .await, + Err(ClientError::Unavailable { .. }) + )); + + let pending_path = temp.path().join("credential/rotation.pending.json"); + let pending = fs::read(&pending_path).expect("read pending state"); + let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read next key"); + let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key"); + let (_, stored) = rotation_response(&pki, &next, 8); + write_stored_credential(&temp.path().join("credential/device.crt.json"), &stored); + + let idle = server(&pki, vec![]).await; + assert!( + client(&idle, &pki, Duration::from_secs(2)) + .rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp()) + .await + .expect("recover after credential save") + .is_none() + ); + assert_eq!( + identity_store + .load() + .expect("load key") + .expect("current key") + .public_key_der(), + next.public_key_der() + ); + + fs::write(&pending_path, pending).expect("restore pending after key commit"); + set_owner_only(&pending_path); + assert!( + client(&idle, &pki, Duration::from_secs(2)) + .rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp()) + .await + .expect("recover after key commit") + .is_none() + ); + assert!(!pending_path.exists()); +} + +#[tokio::test] +async fn pending_reenrollment_blocks_rotation_and_resumes_original_exchange() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let current = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13); + let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + let registered = client(®istration, &pki, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + + let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; + assert!(matches!( + client(&failed, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await, + Err(ClientError::Unavailable { .. }) + )); + + let pending_path = temp.path().join("credential/registration.pending.json"); + let pending = fs::read(&pending_path).expect("read pending reenrollment"); + let pending_document: Value = serde_json::from_slice(&pending).expect("pending JSON"); + let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read next key"); + let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key"); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14); + + let rotation = server(&pki, vec![]).await; + let error = client(&rotation, &pki, Duration::from_secs(2)) + .rotate_if_due(&identity_store, &credential_store, registered.not_after_unix - 8 * 60 * 60) + .await + .expect_err("pending reenrollment blocks rotation"); + assert!(matches!(error, ClientError::PendingRegistration)); + assert!(rotation.seen.lock().expect("seen lock").is_empty()); + + let resumed = server(&pki, vec![Reply::Json(StatusCode::CREATED, enrolled)]).await; + let credential = client(&resumed, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("resume reenrollment"); + assert_eq!(credential.certificate_serial, "0e".repeat(16)); + let resumed_seen = resumed.seen.lock().expect("seen lock"); + assert_eq!(resumed_seen.len(), 1); + assert_eq!(resumed_seen[0]["requestId"], pending_document["requestId"]); + assert_eq!(resumed_seen[0]["certificateRequest"], pending_document["certificateRequest"]); +} + +#[tokio::test] +async fn reenrollment_commit_recovers_after_each_durable_step() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let current = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13); + let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; + client(®istration, &pki, Duration::from_secs(2)) + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + + let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; + assert!(matches!( + client(&failed, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await, + Err(ClientError::Unavailable { .. }) + )); + + let pending_path = temp.path().join("credential/registration.pending.json"); + let pending = fs::read(&pending_path).expect("read pending reenrollment"); + let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read next key"); + let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key"); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14); + write_stored_credential(&temp.path().join("credential/device.crt.json"), &enrolled); + + let idle = server(&pki, vec![]).await; + let idle_client = client(&idle, &pki, Duration::from_secs(2)); + let recovered = idle_client + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("recover after reenrollment credential save"); + assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert_eq!( + identity_store + .load() + .expect("load key") + .expect("current key") + .public_key_der(), + next.public_key_der() + ); + + fs::remove_file(temp.path().join("credential/registration.completed.json")).expect("remove completed receipt"); + fs::write(&pending_path, pending).expect("restore pending after key commit"); + set_owner_only(&pending_path); + let recovered = idle_client + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("recover after reenrollment key commit"); + assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert!(!pending_path.exists()); + let recovered = idle_client + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("completed reenrollment is idempotent after pending cleanup"); + assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert!(idle.seen.lock().expect("seen lock").is_empty()); + + let different = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; + assert!(matches!( + client(&different, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .await, + Err(ClientError::Unavailable { .. }) + )); + assert_eq!(different.seen.lock().expect("seen lock").len(), 3); +} + +#[test] +fn registration_token_schema_is_strict_and_bounded() { + let mut document = serde_json::to_value(token_document()).expect("token document"); + document["unexpected"] = json!(true); + assert!(matches!( + RegistrationToken::from_reader(serde_json::to_vec(&document).expect("token JSON").as_slice()), + Err(TokenError::Invalid(_)) + )); + assert!(matches!( + RegistrationToken::from_reader(vec![b' '; 16 * 1024 + 1].as_slice()), + Err(TokenError::TooLarge) + )); + let mut malformed = token_document(); + malformed["challengeNonce"] = json!("A".repeat(64)); + assert!(matches!( + RegistrationToken::from_reader(serde_json::to_vec(&malformed).expect("token JSON").as_slice()), + Err(TokenError::Shape) + )); +} + +#[tokio::test] +async fn rotation_waits_for_threshold_and_stops_on_revocation() { + let temp = tempfile::tempdir().expect("temp dir"); + let (identity_store, credential_store) = stores(&temp); + let identity = identity_store.load_or_create().expect("create identity"); + let pki = TestPki::new(); + let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 3); + let rotation_server = server( + &pki, + vec![ + Reply::Json(StatusCode::CREATED, issued), + Reply::Json(StatusCode::UNAUTHORIZED, json!({"details": [{"reason": "DEVICE_REVOKED"}]})), + ], + ) + .await; + let connect = client(&rotation_server, &pki, Duration::from_secs(2)); + let current = connect + .register(&identity_store, &credential_store, &token()) + .await + .expect("register"); + assert!( + connect + .rotate_if_due(&identity_store, &credential_store, current.not_before_unix) + .await + .expect("not due") + .is_none() + ); + let error = connect + .rotate_if_due(&identity_store, &credential_store, current.not_after_unix - 8 * 60 * 60) + .await + .expect_err("revocation must stop rotation"); + assert!(matches!(error, ClientError::AccessRevoked { .. })); + assert!(error.to_string().contains("ConnectClient::reenroll")); + assert_eq!(rotation_server.seen.lock().expect("seen lock").len(), 2); + let stored: Value = + serde_json::from_slice(&fs::read(temp.path().join("credential/device.crt.json")).expect("read stored credential")) + .expect("stored credential JSON"); + assert_eq!(stored["certificateSerial"], current.certificate_serial); + + let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read staged next key"); + let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse staged next key"); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 12); + let reenrollment = server(&pki, vec![Reply::Json(StatusCode::CREATED, enrolled)]).await; + let fresh = client(&reenrollment, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("fresh token reenrolls revoked credential"); + assert_eq!(fresh.certificate_serial, "0c".repeat(16)); + assert_eq!( + identity_store + .load() + .expect("load identity") + .expect("identity") + .public_key_der(), + next.public_key_der() + ); +} + +#[test] +fn unconfigured_connect_has_no_side_effects() { + let temp = tempfile::tempdir().expect("temp dir"); + let directory = temp.path().join("connect"); + assert!( + ConnectClient::from_optional_config(None) + .expect("unconfigured is valid") + .is_none() + ); + assert!(!directory.exists()); +}