From 08db857e05dbf1650d56892cfcc0ebd2b1c75c33 Mon Sep 17 00:00:00 2001 From: Leo Galambos Date: Sat, 1 Aug 2026 17:58:16 +0200 Subject: [PATCH] feat(pki): migrate signing authority to transactional metadata Atomically persist signing workflow records and SIGNING_OPERATION content-owner edges through the transactional metadata store. Use the canonical SigningSubmissionId directly as the MetadataKey, remove the former current.bin and signing-specific .owners authority, and recover pending signing operations only from consistent metadata record/owner pairs. Keep immutable staged payload and reference metadata external. Validated: - focused migration tests pass - lib tests pass - PMD and JavaDoc pass - app compilation passes - pki retains only the 31 independently classified credential and revocation failures --- .../pki/impl/core/async/PkiSigningBus.java | 59 +- .../pki/impl/fs/FilesystemPkiStore.java | 510 ++++++++++++++---- .../impl/fs/FilesystemStagedContentStore.java | 139 ++++- .../java/zeroecho/pki/impl/fs/FsPaths.java | 14 +- .../zeroecho/pki/spi/store/MetadataKey.java | 161 +++++- .../fs/FilesystemSignWorkflowStoreTest.java | 94 ++-- .../fs/FilesystemStagedContentStoreTest.java | 32 ++ .../pki/spi/store/MetadataKeyTest.java | 33 +- 8 files changed, 815 insertions(+), 227 deletions(-) diff --git a/pki/src/main/java/zeroecho/pki/impl/core/async/PkiSigningBus.java b/pki/src/main/java/zeroecho/pki/impl/core/async/PkiSigningBus.java index 2ab7c07..93b76be 100644 --- a/pki/src/main/java/zeroecho/pki/impl/core/async/PkiSigningBus.java +++ b/pki/src/main/java/zeroecho/pki/impl/core/async/PkiSigningBus.java @@ -58,7 +58,6 @@ import zeroecho.pki.api.KeyRef; import zeroecho.pki.api.PkiException; import zeroecho.pki.api.PkiId; import zeroecho.pki.api.content.DurableContentReference; -import zeroecho.pki.api.content.DurableContentOwner; import zeroecho.pki.api.audit.Principal; import zeroecho.pki.api.orch.OrchestrationDurabilityPolicy; import zeroecho.pki.api.orch.SigningSubmissionId; @@ -427,36 +426,15 @@ public final class PkiSigningBus implements AutoCloseable { SignWorkflowStore.Record intent = new SignWorkflowStore.Record(baseOpId, namespace, fingerprint, owner, parsed.createdAt(), deadline, persistedRequest, SignWorkflowStore.State.INTENT, 0L, 0L, Optional.empty(), Optional.of("INTENT"), Optional.empty(), Optional.empty()); - DurableContentOwner contentOwner = DurableContentOwner.signingOperation(baseOpId); - boolean retained = false; - boolean recordVisible = false; - try { - retained = store.stagedContent().retainContent(content, contentOwner); - SignWorkflowStore.CreateResult created = store.createSignIntent(intent); - if (created == SignWorkflowStore.CreateResult.CONFLICT) { - throw new PkiException("Signing submission identifier conflicts with a different request"); - } - recordVisible = true; - authoritative = store.getSignRecord(baseOpId).orElseThrow(); - } catch (java.io.IOException exception) { - throw new PkiException("Signing content retention failed: code=SPOOL_STORAGE_FAILED", exception); - } finally { - if (retained && !recordVisible) { - rollbackSigningOwner(content, contentOwner); - } + SignWorkflowStore.CreateResult created = store.createSignIntent(intent); + if (created == SignWorkflowStore.CreateResult.CONFLICT) { + throw new PkiException("Signing submission identifier conflicts with a different request"); } + authoritative = store.getSignRecord(baseOpId).orElseThrow(); } project(authoritative); } - private void rollbackSigningOwner(DurableContentReference content, DurableContentOwner owner) { - try { - store.stagedContent().releaseContent(content, owner); - } catch (java.io.IOException exception) { - throw new PkiException("Signing content rollback failed: code=SPOOL_STORAGE_FAILED", exception); - } - } - /** * Returns current status if known. */ @@ -564,15 +542,8 @@ public final class PkiSigningBus implements AutoCloseable { if (!isTerminalSignState(state.state())) { return; } - Optional releaseReference = Optional.empty(); - if (state.state() != SignWorkflowStore.State.RETIRED) { - releaseReference = Optional.of(SignContinuation.decode(state.request(), store.stagedContent()).content()); - } state = confirmRetirement(baseOpId, state); store.deleteWorkflowState(baseOpId); - if (releaseReference.isPresent()) { - releaseRetiredOperationContent(baseOpId, releaseReference.get()); - } } AsyncState advisoryState = state.result().isPresent() ? AsyncState.SUCCEEDED : AsyncState.CANCELLED; bus.update(baseOpId, new AsyncStatus(advisoryState, store.signingNow(), Optional.of("RETIRED"), @@ -580,16 +551,6 @@ public final class PkiSigningBus implements AutoCloseable { bus.retire(baseOpId); } - private void releaseRetiredOperationContent(PkiId operationId, DurableContentReference reference) { - if (reference.lifecycle() != DurableContentReference.Lifecycle.OPERATION) { - return; - } - try { - store.stagedContent().releaseContent(reference, DurableContentOwner.signingOperation(operationId)); - } catch (java.io.IOException exception) { - throw new PkiException("Signing content retirement failed: code=SPOOL_STORAGE_FAILED", exception); - } - } private void reconcileExpiredOperations() { Instant current = store.signingNow(); @@ -1103,8 +1064,6 @@ public final class PkiSigningBus implements AutoCloseable { boolean reservationTransferred = false; try { SignContinuation continuation = SignContinuation.decode(claimed.request(), store.stagedContent()); - DurableContentOwner contentOwner = DurableContentOwner.signingOperation(opId); - requireSigningOwner(continuation.content(), contentOwner); X509ExecutionPlan plan = authority.planSigning(continuation.algorithmId, workflowImplementationId(signer), SignatureWorkflow.class); authority.authorize(plan, signer, AlgorithmExecutionCapability.Direction.SIGN); @@ -1133,16 +1092,6 @@ public final class PkiSigningBus implements AutoCloseable { } } - private void requireSigningOwner(DurableContentReference reference, DurableContentOwner owner) { - try { - if (!store.stagedContent().contentOwners(reference).contains(owner)) { - throw new PkiException("Signing content owner missing: code=STAGED_CONTENT_INCOMPLETE"); - } - } catch (java.io.IOException exception) { - throw new PkiException("Signing content ownership failed: code=CONTENT_INTEGRITY_FAILED", exception); - } - } - private RepeatableContent openContent(DurableContentReference reference) { try { return store.stagedContent().openContent(reference); diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java index ca6d3f1..f4e3f3b 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java @@ -34,7 +34,9 @@ package zeroecho.pki.impl.fs; import java.io.Closeable; +import java.io.ByteArrayInputStream; import java.io.IOException; +import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.channels.FileLock; @@ -55,10 +57,12 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.HexFormat; +import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.OptionalLong; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -95,6 +99,14 @@ import zeroecho.pki.impl.ProfileLifecycleFailure; import zeroecho.pki.impl.ProfileLifecycleFailure.Code; import zeroecho.pki.impl.core.async.PkiSigningBus; import zeroecho.pki.spi.store.PkiStore; +import zeroecho.core.io.CancellationSignal; +import zeroecho.core.io.RepeatableContent; +import zeroecho.pki.spi.store.MetadataCommitResult; +import zeroecho.pki.spi.store.MetadataCursor; +import zeroecho.pki.spi.store.MetadataKey; +import zeroecho.pki.spi.store.MetadataSnapshot; +import zeroecho.pki.spi.store.MetadataStoreId; +import zeroecho.pki.spi.store.MetadataTransaction; import zeroecho.pki.spi.store.StagedContentStore; import zeroecho.pki.spi.store.SignWorkflowStore; import zeroecho.pki.spi.store.TemporaryUniqueIndex; @@ -155,9 +167,10 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { private static final Logger LOG = Logger.getLogger(FilesystemPkiStore.class.getName()); /* package */ static final String CURRENT_STORE_VERSION = "v2"; - private static final int SIGN_RECORD_MAGIC = 0x5A455352; + private static final String SIGN_RECORD_NAMESPACE = "io.zeroecho.pki.signing-record"; + private static final String SIGN_OWNER_NAMESPACE = "io.zeroecho.pki.signing-owner"; private static final int CURRENT_SIGN_RECORD_VERSION = 2; - private static final int SIGN_RECORD_HEADER_BYTES = Integer.BYTES * 2; + private static final int SIGN_OWNER_VALUE_VERSION = 1; private static final String SIGN_FINGERPRINT_PREFIX = "signfp:v1:"; private static final int SIGN_FINGERPRINT_HEX_LENGTH = 64; private static final long INITIAL_FENCE = 0L; @@ -176,6 +189,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { private final AtomicBoolean durabilityUncertain; private final FilesystemStagedContentStore stagedContent; private final CredentialContentTransaction credentialContentTransactions; + private final PosixTransactionalMetadataStore metadataStore; private final StoreOwnership ownership; @@ -227,11 +241,16 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } boolean ownershipTransferred = false; + PosixTransactionalMetadataStore openedMetadata = null; try { ensureVersionFile(); this.signingNamespace = ensureSigningNamespace(); this.stagedContent = new FilesystemStagedContentStore(this.paths.stagedContentRoot(), this.signingNamespace); + FsOperations.ensureDir(this.paths.transactionalMetadataLog().getParent()); + openedMetadata = openMetadataStore(); + this.metadataStore = openedMetadata; + this.stagedContent.bindSigningOwnership(this::findSigningOwner); this.credentialContentTransactions = new CredentialContentTransaction(this.paths, this.stagedContent); this.signingTimeWatermark = new AtomicLong(loadSigningTimeWatermark()); this.historySeq = new AtomicLong(0L); @@ -244,11 +263,66 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { throw new IllegalStateException("failed to open filesystem store at " + root, e); } finally { if (!ownershipTransferred) { + if (openedMetadata != null) { + try { + closeMetadataAfterFailedOpen(openedMetadata); + } catch (IOException closeFailure) { + LOG.log(Level.WARNING, "Metadata-store cleanup failed during initialization"); + } + } acquiredOwnership.closeSilently(); } } } + private static void closeMetadataAfterFailedOpen(PosixTransactionalMetadataStore openedMetadata) + throws IOException { + openedMetadata.close(); + } + + private PosixTransactionalMetadataStore openMetadataStore() throws IOException { + Path log = paths.transactionalMetadataLog(); + PosixTransactionalMetadataStore opened = Files.exists(log) + ? PosixTransactionalMetadataStore.open(log) + : PosixTransactionalMetadataStore.create(log, new MetadataStoreId(signingNamespace)); + if (!opened.id().equals(new MetadataStoreId(signingNamespace))) { + try { + opened.close(); + } catch (IOException closeFailure) { + throw new IOException("Metadata store authority mismatch", closeFailure); + } + throw new IOException("Metadata store authority mismatch"); + } + return opened; + } + + private Set findSigningOwner(DurableContentReference reference) throws IOException { + Set owners = new HashSet<>(); + try (MetadataSnapshot snapshot = metadataStore.snapshot(); + MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(SIGN_OWNER_NAMESPACE), + CancellationSignal.NONE)) { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + MetadataSnapshot.Record owner = next.orElseThrow(); + try (owner) { + Optional matched = matchingSigningOwner(owner, reference); + matched.ifPresent(owners::add); + } + } + return Set.copyOf(owners); + } + } + + private Optional matchingSigningOwner(MetadataSnapshot.Record owner, + DurableContentReference reference) + throws IOException { + PkiId submissionId = new PkiId(owner.key().key()); + if (reference.equals(decodeSigningOwner(owner, submissionId))) { + return Optional.of(DurableContentOwner.signingOperation(submissionId)); + } + return Optional.empty(); + } + @Override public StagedContentStore stagedContent() { return stagedContent; @@ -261,7 +335,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { try { addPersistedCredentialReferences(retained, retainedOwners); addPersistedStatusReferences(retained); - addPendingSigningReferences(retained, retainedOwners); + addPendingSigningReferences(retained); stagedContent.recoverContent(retained, retainedOwners); } catch (IllegalStateException | PkiException malformedDurableState) { // Recovery cannot prove abandonment while durable metadata is @@ -307,34 +381,10 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } } - private void addPendingSigningReferences(TemporaryUniqueIndex retained, TemporaryUniqueIndex retainedOwners) - throws IOException { - Path root = paths.signWorkflowRoot(); - if (!Files.isDirectory(root)) { - return; - } - try (Stream pathsStream = Files.walk(root)) { - java.util.Iterator iterator = pathsStream - .filter(path -> Files.isRegularFile(path) - && FsPaths.CURRENT_FILE.equals(path.getFileName().toString())) - .iterator(); - while (iterator.hasNext()) { - SignWorkflowStore.Record record = readSignRecordFile(iterator.next()); - PkiSigningBus.SignContinuation continuation = PkiSigningBus.SignContinuation.decode(record.request(), - stagedContent); - if (record.state() == SignWorkflowStore.State.RETIRED) { - if (continuation.hasLiveContent()) { - throw new IllegalStateException("Retired signing record retains live content"); - } - continue; - } - DurableContentReference reference = continuation.content(); - DurableContentOwner owner = DurableContentOwner.signingOperation(record.submissionId()); - if (!stagedContent.contentOwners(reference).contains(owner)) { - throw new IllegalStateException("Signing content owner is missing"); - } - addRetained(retained, reference); - retainedOwners.add(owner.canonicalForm().getBytes(StandardCharsets.UTF_8)); + private void addPendingSigningReferences(TemporaryUniqueIndex retained) throws IOException { + for (StoredSign stored : listStoredSigns()) { + if (stored.record().state() != SignWorkflowStore.State.RETIRED) { + addRetained(retained, stored.reference().orElseThrow()); } } } @@ -812,13 +862,43 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { ? SignWorkflowStore.CreateResult.ATTACHED : SignWorkflowStore.CreateResult.CONFLICT; } - writeSignRecord(intent); - return SignWorkflowStore.CreateResult.CREATED; + PkiSigningBus.SignContinuation continuation = PkiSigningBus.SignContinuation.decode(intent.request(), + stagedContent); + DurableContentReference reference = continuation.content(); + MetadataCommitResult result; + try (FilesystemStagedContentStore.SigningReservation ignoredReservation = + stagedContent.reserveSigningPublication(reference); + RepeatableContent ignoredContent = stagedContent.openContent(reference)) { + stagedContent.restoreReference(reference.storeId(), reference.contentId(), reference.encoding(), + reference.length(), reference.sha256(), reference.lifecycle()); + result = createSignMetadata(intent, reference); + } catch (IOException exception) { + retireFailedSigningPublication(reference); + throw new PkiException("Signing staged content is invalid: code=CONTENT_INTEGRITY_FAILED", exception); + } catch (PkiException | IllegalArgumentException | IllegalStateException exception) { + if (!durabilityUncertain.get()) { + retireFailedSigningPublication(reference); + } + throw exception; + } + if (result.outcome() == MetadataCommitResult.Outcome.COMMITTED) { + return SignWorkflowStore.CreateResult.CREATED; + } + retireFailedSigningPublication(reference); + return SignWorkflowStore.CreateResult.CONFLICT; } finally { releaseSignLock(intent.submissionId(), lock); } } + private void retireFailedSigningPublication(DurableContentReference reference) { + try { + stagedContent.retireUnownedContent(reference); + } catch (IOException cleanupFailure) { + LOG.log(Level.WARNING, "Failed signing publication cleanup was incomplete"); + } + } + @Override public Optional getSignRecord(PkiId submissionId) { requireStoreUsable(); @@ -834,15 +914,32 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { @Override public List listSignRecords() { requireStoreUsable(); - Path root = paths.signWorkflowRoot(); - if (!Files.isDirectory(root)) { - return List.of(); - } - try (Stream directories = Files.list(root)) { - return directories.filter(Files::isDirectory).map(directory -> directory.resolve(FsPaths.CURRENT_FILE)) - .filter(Files::isRegularFile).sorted(Comparator.comparing(Path::toString)) - .map(this::readSignRecordFile).peek(record -> validateSignRecord(record.submissionId(), record)) - .toList(); + return listStoredSigns().stream().map(StoredSign::record).toList(); + } + + private List listStoredSigns() { + List storedSigns = new ArrayList<>(); + Set recordIdentities = new HashSet<>(); + try (MetadataSnapshot snapshot = metadataStore.snapshot()) { + try (MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(SIGN_RECORD_NAMESPACE), + CancellationSignal.NONE)) { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + StoredSign stored = decodeStoredSign(snapshot, next.orElseThrow()); + storedSigns.add(stored); + recordIdentities.add(stored.record().submissionId().value()); + } + } + try (MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(SIGN_OWNER_NAMESPACE), + CancellationSignal.NONE)) { + Optional next; + while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) { + if (!recordIdentities.contains(next.orElseThrow().key().key())) { + throw new IllegalStateException("Signing owner exists without its record"); + } + } + } + return List.copyOf(storedSigns); } catch (IOException ex) { throw new IllegalStateException("Failed to list authoritative signing records", ex); } @@ -854,11 +951,12 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { requirePositive(lease, "lease"); SignLockEntry lock = acquireSignLock(submissionId); try { - Optional optional = readSignRecord(submissionId); + Optional optional = readStoredSign(submissionId); if (optional.isEmpty()) { return Optional.empty(); } - SignWorkflowStore.Record current = optional.get(); + StoredSign stored = optional.get(); + SignWorkflowStore.Record current = stored.record(); Instant now = signingNow(); boolean claimable = current.state() == SignWorkflowStore.State.INTENT; boolean leaseAvailable = current.leaseUntil().isEmpty() || !current.leaseUntil().get().isAfter(now); @@ -868,8 +966,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { SignWorkflowStore.Record claimed = copySignRecord(current, current.state(), current.revision() + 1L, current.fence() + 1L, Optional.of(now.plus(lease)), current.detailCode(), current.result(), current.providerUpdatedAt()); - writeSignRecord(claimed); - return Optional.of(claimed); + return replaceSignMetadata(stored, claimed, false) ? Optional.of(claimed) : Optional.empty(); } finally { releaseSignLock(submissionId, lock); } @@ -882,19 +979,19 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { requirePositive(lease, "lease"); SignLockEntry lock = acquireSignLock(submissionId); try { - Optional optional = readSignRecord(submissionId); + Optional optional = readStoredSign(submissionId); if (optional.isEmpty()) { return Optional.empty(); } - SignWorkflowStore.Record current = optional.get(); + StoredSign stored = optional.get(); + SignWorkflowStore.Record current = stored.record(); if (current.revision() != expectedRevision || current.fence() != fence || current.leaseUntil().isEmpty()) { return Optional.empty(); } SignWorkflowStore.Record renewed = copySignRecord(current, current.state(), current.revision() + 1L, fence, Optional.of(signingNow().plus(lease)), current.detailCode(), current.result(), current.providerUpdatedAt()); - writeSignRecord(renewed); - return Optional.of(renewed); + return replaceSignMetadata(stored, renewed, false) ? Optional.of(renewed) : Optional.empty(); } finally { releaseSignLock(submissionId, lock); } @@ -911,11 +1008,12 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { Objects.requireNonNull(providerUpdatedAt, "providerUpdatedAt"); SignLockEntry lock = acquireSignLock(submissionId); try { - Optional optional = readSignRecord(submissionId); + Optional optional = readStoredSign(submissionId); if (optional.isEmpty()) { return Optional.empty(); } - SignWorkflowStore.Record current = optional.get(); + StoredSign stored = optional.get(); + SignWorkflowStore.Record current = stored.record(); if (current.revision() != expectedRevision || current.fence() != fence || !validSignTransition(current.state(), target)) { return Optional.empty(); @@ -930,8 +1028,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } SignWorkflowStore.Record transitioned = copySignRecord(current, target, current.revision() + 1L, fence, Optional.empty(), detailCode, result, providerUpdatedAt); - writeSignRecord(transitioned); - return Optional.of(transitioned); + return replaceSignMetadata(stored, transitioned, false) ? Optional.of(transitioned) : Optional.empty(); } finally { releaseSignLock(submissionId, lock); } @@ -942,11 +1039,12 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { requireStoreUsable(); SignLockEntry lock = acquireSignLock(submissionId); try { - Optional optional = readSignRecord(submissionId); + Optional optional = readStoredSign(submissionId); if (optional.isEmpty()) { return Optional.empty(); } - SignWorkflowStore.Record current = optional.get(); + StoredSign stored = optional.get(); + SignWorkflowStore.Record current = stored.record(); if (current.revision() != expectedRevision || current.fence() != fence || !isRetirableSignState(current.state())) { return Optional.empty(); @@ -958,7 +1056,18 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { current.namespace(), current.fingerprint(), current.owner(), current.createdAt(), current.deadline(), retiredRequest, SignWorkflowStore.State.RETIRED, current.revision() + 1L, fence, Optional.empty(), Optional.of("RETIRED"), current.result(), current.providerUpdatedAt()); - writeSignRecord(retired); + if (!replaceSignMetadata(stored, retired, true)) { + return Optional.empty(); + } + stored.reference().ifPresent(reference -> { + try { + if (reference.lifecycle() == DurableContentReference.Lifecycle.OPERATION) { + stagedContent.retireSigningContent(reference); + } + } catch (IOException exception) { + LOG.log(Level.WARNING, "Signing content retirement cleanup failed"); + } + }); return Optional.of(retired); } finally { releaseSignLock(submissionId, lock); @@ -968,24 +1077,19 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { @Override public int purgeExpiredSignRecords() { requireStoreUsable(); - Path root = paths.signWorkflowRoot(); - if (!Files.isDirectory(root)) { - return 0; - } int purged = 0; for (SignWorkflowStore.Record record : listSignRecords()) { if (record.state() == SignWorkflowStore.State.RETIRED && !record.createdAt().plus(options.signingOperationHorizon()).isAfter(signingNow())) { SignLockEntry lock = acquireSignLock(record.submissionId()); try { - Optional current = readSignRecord(record.submissionId()); - if (current.isPresent() && current.get().state() == SignWorkflowStore.State.RETIRED - && !current.get().createdAt().plus(options.signingOperationHorizon()).isAfter(signingNow()) - && Files.deleteIfExists(paths.signWorkflowPath(record.submissionId()))) { + Optional current = readStoredSign(record.submissionId()); + if (current.isPresent() && current.get().record().state() == SignWorkflowStore.State.RETIRED + && !current.get().record().createdAt().plus(options.signingOperationHorizon()) + .isAfter(signingNow()) + && deleteSignMetadata(current.get())) { purged++; } - } catch (IOException ex) { - throw new IllegalStateException("failed to purge signing workflow", ex); } finally { releaseSignLock(record.submissionId(), lock); } @@ -996,7 +1100,24 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { @Override public void close() throws IOException { - this.ownership.close(); + IOException failure = null; + try { + metadataStore.close(); + } catch (IOException exception) { + failure = exception; + } + try { + ownership.close(); + } catch (IOException exception) { + if (failure == null) { + failure = exception; + } else { + failure.addSuppressed(exception); + } + } + if (failure != null) { + throw failure; + } } private void requireStoreUsable() { @@ -1351,36 +1472,57 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { } private Optional readSignRecord(PkiId submissionId) { - Path path = paths.signWorkflowPath(submissionId); - if (!Files.exists(path)) { - return Optional.empty(); - } - SignWorkflowStore.Record record = readSignRecordFile(path); - validateSignRecord(submissionId, record); - return Optional.of(record); + return readStoredSign(submissionId).map(StoredSign::record); } - private SignWorkflowStore.Record readSignRecordFile(Path path) { - try { - byte[] envelope = FsOperations.readAll(path); - if (envelope.length < SIGN_RECORD_HEADER_BYTES) { - throw new IllegalStateException("Invalid signing record envelope"); + private Optional readStoredSign(PkiId submissionId) { + MetadataKey recordKey = signingRecordKey(submissionId); + try (MetadataSnapshot snapshot = metadataStore.snapshot()) { + Optional storedRecord = snapshot.get(recordKey); + if (storedRecord.isEmpty()) { + if (snapshot.get(signingOwnerKey(submissionId)).isPresent()) { + throw new IllegalStateException("Signing owner exists without its record"); + } + return Optional.empty(); } - ByteBuffer input = ByteBuffer.wrap(envelope); - if (input.getInt() != SIGN_RECORD_MAGIC) { - throw new IllegalStateException("Invalid signing record envelope"); - } - if (input.getInt() != CURRENT_SIGN_RECORD_VERSION) { - throw new IllegalStateException("Unsupported signing record version"); - } - byte[] payload = new byte[input.remaining()]; - input.get(payload); - return FsCodec.decode(FsCodec.SIGN_WORKFLOW_RECORD, payload, stagedContent); + return Optional.of(decodeStoredSign(snapshot, storedRecord.orElseThrow())); } catch (IOException ex) { throw new IllegalStateException("Failed to read authoritative signing record", ex); } } + private StoredSign decodeStoredSign(MetadataSnapshot snapshot, MetadataSnapshot.Record storedRecord) + throws IOException { + SignWorkflowStore.Record record = decodeSignRecord(storedRecord); + PkiId submissionId = record.submissionId(); + if (!signingRecordKey(submissionId).equals(storedRecord.key())) { + throw new IllegalStateException("Signing record metadata key mismatch"); + } + validateSignRecord(submissionId, record); + Optional storedOwner = snapshot.get(signingOwnerKey(submissionId)); + if (record.state() == SignWorkflowStore.State.RETIRED) { + if (storedOwner.isPresent()) { + throw new IllegalStateException("Retired signing record retains an owner edge"); + } + return new StoredSign(record, storedRecord.recordRevision(), OptionalLong.empty(), Optional.empty()); + } + try (MetadataSnapshot.Record owner = storedOwner.orElseThrow( + () -> new IllegalStateException("Signing record owner edge is missing"))) { + DurableContentReference reference = decodeSigningOwner(owner, submissionId); + PkiSigningBus.SignContinuation continuation = PkiSigningBus.SignContinuation.decode( + record.request(), stagedContent); + if (!reference.equals(continuation.content())) { + throw new IllegalStateException("Signing record owner reference mismatch"); + } + return new StoredSign(record, storedRecord.recordRevision(), OptionalLong.of(owner.recordRevision()), + Optional.of(reference)); + } + } + + private SignWorkflowStore.Record decodeSignRecord(MetadataSnapshot.Record stored) throws IOException { + return FsCodec.decode(FsCodec.SIGN_WORKFLOW_RECORD, readMetadataValue(stored), stagedContent); + } + // Parsing and canonicalization failures can carry persisted request material; // validation deliberately replaces every such cause with a safe corruption // code. @@ -1456,20 +1598,53 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { validateSignRecordState(record); } - private void writeSignRecord(SignWorkflowStore.Record record) { + private MetadataCommitResult createSignMetadata(SignWorkflowStore.Record record, + DurableContentReference reference) { validateSignRecord(record.submissionId(), record); - try { - byte[] payload = FsCodec.encode(FsCodec.SIGN_WORKFLOW_RECORD, record); - ByteBuffer envelope = ByteBuffer.allocate(SIGN_RECORD_HEADER_BYTES + payload.length); - envelope.putInt(SIGN_RECORD_MAGIC); - envelope.putInt(CURRENT_SIGN_RECORD_VERSION); - envelope.put(payload); - FsOperations.writeAtomic(paths.signWorkflowPath(record.submissionId()), envelope.array()); + try (MetadataTransaction transaction = metadataStore.beginTransaction()) { + transaction.create(signingRecordKey(record.submissionId()), + byteContent(FsCodec.encode(FsCodec.SIGN_WORKFLOW_RECORD, record)), CancellationSignal.NONE); + transaction.create(signingOwnerKey(record.submissionId()), + byteContent(encodeSigningOwner(record.submissionId(), reference)), CancellationSignal.NONE); + return requireKnownOutcome(transaction.commit()); } catch (IOException ex) { - throw new IllegalStateException("failed to persist signing workflow", ex); + throw new IllegalStateException("Failed to persist signing authority", ex); } } + private boolean replaceSignMetadata(StoredSign current, SignWorkflowStore.Record replacement, + boolean removeOwner) { + validateSignRecord(replacement.submissionId(), replacement); + try (MetadataTransaction transaction = metadataStore.beginTransaction()) { + transaction.replace(signingRecordKey(replacement.submissionId()), current.recordRevision(), + byteContent(FsCodec.encode(FsCodec.SIGN_WORKFLOW_RECORD, replacement)), CancellationSignal.NONE); + if (removeOwner) { + transaction.delete(signingOwnerKey(replacement.submissionId()), + current.ownerRevision().orElseThrow()); + } + return requireKnownOutcome(transaction.commit()).outcome() == MetadataCommitResult.Outcome.COMMITTED; + } catch (IOException ex) { + throw new IllegalStateException("Failed to replace signing authority", ex); + } + } + + private boolean deleteSignMetadata(StoredSign current) { + try (MetadataTransaction transaction = metadataStore.beginTransaction()) { + transaction.delete(signingRecordKey(current.record().submissionId()), current.recordRevision()); + return requireKnownOutcome(transaction.commit()).outcome() == MetadataCommitResult.Outcome.COMMITTED; + } catch (IOException ex) { + throw new IllegalStateException("Failed to retire signing authority", ex); + } + } + + private MetadataCommitResult requireKnownOutcome(MetadataCommitResult result) { + if (result.outcome() == MetadataCommitResult.Outcome.UNKNOWN) { + durabilityUncertain.set(true); + throw new PkiException("Store durability unconfirmed: code=STORE_DURABILITY_UNCONFIRMED"); + } + return result; + } + private static SignWorkflowStore.Record copySignRecord(SignWorkflowStore.Record current, SignWorkflowStore.State state, long revision, long fence, Optional leaseUntil, Optional detailCode, Optional result, Optional providerUpdatedAt) { @@ -1478,6 +1653,143 @@ public final class FilesystemPkiStore implements PkiStore, Closeable { leaseUntil, detailCode, result, providerUpdatedAt); } + private static MetadataKey signingRecordKey(PkiId submissionId) { + SigningSubmissionId parsed = SigningSubmissionId.parse(submissionId); + return new MetadataKey(SIGN_RECORD_NAMESPACE, parsed.id().value()); + } + + private static MetadataKey signingOwnerKey(PkiId submissionId) { + SigningSubmissionId parsed = SigningSubmissionId.parse(submissionId); + return new MetadataKey(SIGN_OWNER_NAMESPACE, parsed.id().value()); + } + + private static RepeatableContent byteContent(byte[] value) { + return new ByteValueContent(value); + } + + private static byte[] readMetadataValue(MetadataSnapshot.Record record) throws IOException { + long length = record.length().orElseThrow(); + if (length < 0L || length > FsCodec.MAX_COMPONENT_BYTES) { + throw new IOException("Signing metadata value length is invalid"); + } + byte[] result = new byte[Math.toIntExact(length)]; + try (InputStream input = record.openStream()) { + int offset = 0; + while (offset < result.length) { + int count = input.read(result, offset, result.length - offset); + if (count <= 0) { + throw new IOException("Signing metadata value is truncated"); + } + offset += count; + } + if (input.read() >= 0) { + throw new IOException("Signing metadata value has trailing data"); + } + } + return result; + } + + private static byte[] encodeSigningOwner(PkiId submissionId, DurableContentReference reference) { + byte[][] fields = { + DurableContentOwner.Category.SIGNING_OPERATION.name().getBytes(StandardCharsets.US_ASCII), + submissionId.value().getBytes(StandardCharsets.UTF_8), + reference.storeId().getBytes(StandardCharsets.UTF_8), + reference.contentId().getBytes(StandardCharsets.UTF_8), + reference.encoding().name().getBytes(StandardCharsets.US_ASCII), + reference.sha256().getBytes(StandardCharsets.US_ASCII), + reference.lifecycle().name().getBytes(StandardCharsets.US_ASCII) + }; + int size = Integer.BYTES + Long.BYTES; + for (byte[] field : fields) { + size = Math.addExact(size, Math.addExact(Integer.BYTES, field.length)); + } + ByteBuffer output = ByteBuffer.allocate(size); + output.putInt(SIGN_OWNER_VALUE_VERSION); + for (byte[] field : fields) { + output.putInt(field.length).put(field); + } + output.putLong(reference.length()); + return output.array(); + } + + private DurableContentReference decodeSigningOwner(MetadataSnapshot.Record owner, PkiId expectedId) + throws IOException { + ByteBuffer input = ByteBuffer.wrap(readMetadataValue(owner)); + try { + if (input.getInt() != SIGN_OWNER_VALUE_VERSION) { + throw new IOException("Unsupported signing owner metadata"); + } + DurableContentOwner.Category category = DurableContentOwner.Category.valueOf(readOwnerField(input)); + String ownerId = readOwnerField(input); + String storeId = readOwnerField(input); + String contentId = readOwnerField(input); + zeroecho.pki.api.Encoding encoding = zeroecho.pki.api.Encoding.valueOf(readOwnerField(input)); + String digest = readOwnerField(input); + DurableContentReference.Lifecycle lifecycle = DurableContentReference.Lifecycle.valueOf( + readOwnerField(input)); + long length = input.getLong(); + if (input.hasRemaining() || category != DurableContentOwner.Category.SIGNING_OPERATION + || !ownerId.equals(expectedId.value())) { + throw new IOException("Signing owner metadata identity mismatch"); + } + DurableContentReference restored = stagedContent.restoreReference(storeId, contentId, encoding, + length, digest, lifecycle); + if (!signingOwnerKey(expectedId).equals(owner.key())) { + throw new IOException("Signing owner metadata key mismatch"); + } + return restored; + } catch (IllegalArgumentException | java.nio.BufferUnderflowException exception) { + throw new IOException("Malformed signing owner metadata", exception); + } + } + + private static String readOwnerField(ByteBuffer input) throws IOException { + int length = input.getInt(); + if (length <= 0 || length > MetadataKey.MAXIMUM_KEY_UTF8_BYTES || input.remaining() < length) { + throw new IOException("Malformed signing owner metadata field"); + } + byte[] value = new byte[length]; + input.get(value); + String decoded = new String(value, StandardCharsets.UTF_8); + if (!Arrays.equals(value, decoded.getBytes(StandardCharsets.UTF_8))) { + throw new IOException("Malformed signing owner metadata UTF-8"); + } + return decoded; + } + + private record StoredSign(SignWorkflowStore.Record record, long recordRevision, + OptionalLong ownerRevision, Optional reference) { + } + + /** Store-owned bounded finite control metadata used only for synchronous admission. */ + private static final class ByteValueContent implements RepeatableContent { + private final byte[] value; + + private ByteValueContent(byte[] value) { + this.value = Objects.requireNonNull(value, "value").clone(); + } + + @Override + public InputStream openStream() { + return new ByteArrayInputStream(value); + } + + @Override + public OptionalLong length() { + return OptionalLong.of(value.length); + } + + @Override + public String contentId() { + return "signing-control-metadata"; + } + + @Override + public void close() { + Arrays.fill(value, (byte) 0); + } + } + private static boolean validSignTransition(SignWorkflowStore.State source, SignWorkflowStore.State target) { if (source == SignWorkflowStore.State.INTENT) { return target == SignWorkflowStore.State.DISPATCHED || target == SignWorkflowStore.State.FAILED diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemStagedContentStore.java b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemStagedContentStore.java index 8baabfd..f4d3a4d 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemStagedContentStore.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FilesystemStagedContentStore.java @@ -61,7 +61,10 @@ import java.util.Objects; import java.util.OptionalLong; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import zeroecho.core.io.RepeatableContent; @@ -100,6 +103,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore { private final Path root; private final String storeId; private final ReentrantLock[] ownerLocks; + private final ConcurrentMap signingReservations; + private final AtomicReference signingOwnership; /** * Creates a staged-content store. @@ -113,6 +118,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore { this.root = Objects.requireNonNull(root, "root").toAbsolutePath().normalize(); this.storeId = requireStoreIdentifier(storeId); this.ownerLocks = new ReentrantLock[OWNER_LOCK_COUNT]; + this.signingReservations = new ConcurrentHashMap<>(); + this.signingOwnership = new AtomicReference<>(reference -> Set.of()); for (int index = 0; index < ownerLocks.length; index++) { ownerLocks[index] = new ReentrantLock(); } @@ -177,6 +184,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore { public boolean retainContent(DurableContentReference reference, DurableContentOwner owner) throws IOException { DurableContentReference exact = requireOwned(reference); Objects.requireNonNull(owner, "owner"); + SigningOwnershipIo.requireSidecarOwner(owner); ReentrantLock lock = ownerLock(exact.contentId()); lock.lock(); try { @@ -196,6 +204,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore { public boolean releaseContent(DurableContentReference reference, DurableContentOwner owner) throws IOException { DurableContentReference exact = requireOwned(reference); Objects.requireNonNull(owner, "owner"); + SigningOwnershipIo.requireSidecarOwner(owner); ReentrantLock lock = ownerLock(exact.contentId()); lock.lock(); try { @@ -208,9 +217,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore { return false; } writeOwners(exact.contentId(), owners); - if (owners.isEmpty()) { - retireFiles(exact); - } + SigningOwnershipIo.retireIfUnowned(this, exact, owners); return true; } finally { lock.unlock(); @@ -224,7 +231,9 @@ public final class FilesystemStagedContentStore implements StagedContentStore { lock.lock(); try { requireExactMetadata(exact); - return Set.copyOf(readOwners(exact.contentId())); + Set owners = readOwners(exact.contentId()); + owners.addAll(signingOwnership.get().findOwners(exact)); + return Set.copyOf(owners); } finally { lock.unlock(); } @@ -237,9 +246,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore { lock.lock(); try { requireExactMetadata(exact); - if (!readOwners(exact.contentId()).isEmpty()) { - throw new IOException("Staged content remains durably owned"); - } + SigningOwnershipIo.requireUnowned(this, exact); retireFiles(exact); } finally { lock.unlock(); @@ -309,9 +316,55 @@ public final class FilesystemStagedContentStore implements StagedContentStore { } private void writeOwners(String contentId, Set owners) throws IOException { + SigningOwnershipIo.requireSidecarOwners(owners); StoreIo.writeOwners(this, contentId, owners); } + /* package */ void retireSigningContent(DurableContentReference reference) throws IOException { + SigningOwnershipIo.retireSigningContent(this, reference); + } + + /* package */ void bindSigningOwnership(SigningOwnership ownership) { + signingOwnership.set(Objects.requireNonNull(ownership, "ownership")); + } + + /* package */ SigningReservation reserveSigningPublication(DurableContentReference reference) + throws IOException { + return SigningOwnershipIo.reserveSigningPublication(this, reference); + } + + /** Internal callback to the transactional signing-owner authority. */ + /* default */ + @FunctionalInterface + interface SigningOwnership { + /** Finds every exact signing owner for a durable content reference. */ + Set findOwners(DurableContentReference reference) throws IOException; + } + + /** Short-lived reference-counted reservation spanning validation and commit. */ + /* package */ final class SigningReservation implements AutoCloseable { + private final String contentId; + private final AtomicBoolean closed = new AtomicBoolean(); + + private SigningReservation(String contentId) { + this.contentId = contentId; + } + + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + ReentrantLock lock = ownerLock(contentId); + lock.lock(); + try { + signingReservations.computeIfPresent(contentId, + (ignored, count) -> count == 1 ? null : count - 1); + } finally { + lock.unlock(); + } + } + } + } + private void retireFiles(DurableContentReference reference) throws IOException { Files.deleteIfExists(metadataPath(reference.contentId())); Files.deleteIfExists(completePath(reference)); @@ -810,6 +863,9 @@ public final class FilesystemStagedContentStore implements StagedContentStore { } for (int index = 0; index < count; index++) { DurableContentOwner owner = parseOwner(input.readUTF()); + if (owner.category() == DurableContentOwner.Category.SIGNING_OPERATION) { + continue; + } if (!owners.add(owner)) { throw new IOException("Duplicate staged content owner"); } @@ -877,6 +933,75 @@ public final class FilesystemStagedContentStore implements StagedContentStore { } } + /** Isolates transactional signing-ownership lifecycle decisions. */ + private static final class SigningOwnershipIo { + private static void requireSidecarOwners(Set owners) { + for (DurableContentOwner owner : owners) { + requireSidecarOwner(owner); + } + } + + private static void requireSidecarOwner(DurableContentOwner owner) { + if (owner.category() == DurableContentOwner.Category.SIGNING_OPERATION) { + throw new IllegalArgumentException( + "Signing-operation ownership is authoritative transactional metadata"); + } + } + + private static SigningReservation reserveSigningPublication(FilesystemStagedContentStore store, + DurableContentReference reference) throws IOException { + DurableContentReference exact = store.requireOwned(reference); + ReentrantLock lock = store.ownerLock(exact.contentId()); + lock.lock(); + try { + store.requireExactMetadata(exact); + store.signingReservations.merge(exact.contentId(), 1, Math::addExact); + return store.new SigningReservation(exact.contentId()); + } catch (ArithmeticException exception) { + throw new IOException("Signing content reservation limit exceeded", exception); + } finally { + lock.unlock(); + } + } + + private static boolean hasSigningAuthority(FilesystemStagedContentStore store, + DurableContentReference reference) throws IOException { + return store.signingReservations.containsKey(reference.contentId()) + || !store.signingOwnership.get().findOwners(reference).isEmpty(); + } + + private static void retireIfUnowned(FilesystemStagedContentStore store, + DurableContentReference reference, Set owners) throws IOException { + if (owners.isEmpty() && !hasSigningAuthority(store, reference)) { + store.retireFiles(reference); + } + } + + private static void requireUnowned(FilesystemStagedContentStore store, + DurableContentReference reference) throws IOException { + if (!store.readOwners(reference.contentId()).isEmpty() + || hasSigningAuthority(store, reference)) { + throw new IOException("Staged content remains durably owned"); + } + } + + private static void retireSigningContent(FilesystemStagedContentStore store, + DurableContentReference reference) throws IOException { + DurableContentReference exact = store.requireOwned(reference); + ReentrantLock lock = store.ownerLock(exact.contentId()); + lock.lock(); + try { + store.requireExactMetadata(exact); + if (store.readOwners(exact.contentId()).isEmpty() + && !hasSigningAuthority(store, exact)) { + store.retireFiles(exact); + } + } finally { + lock.unlock(); + } + } + } + private record StoreReference(String storeId, String contentId, Encoding encoding, long length, String sha256, DurableContentReference.Lifecycle lifecycle) implements DurableContentReference { private StoreReference { diff --git a/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java b/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java index 87247b9..88efa11 100644 --- a/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java +++ b/pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java @@ -94,6 +94,10 @@ final class FsPaths { return this.root.resolve("staged-content"); } + /* default */ Path transactionalMetadataLog() { + return this.root.resolve("metadata").resolve("transactions.log"); + } + /* default */ Path lockFile() { return this.root.resolve(LOCK_DIR).resolve(STORE_LOCK); } @@ -156,16 +160,6 @@ final class FsPaths { return this.root.resolve("requests").resolve(BY_ID).resolve(FsUtil.safeId(requestId) + BINARY_EXTENSION); } - /* default */ Path signWorkflowPath(final PkiId submissionId) { - Objects.requireNonNull(submissionId, "submissionId"); - return this.root.resolve("sign-workflows").resolve(BY_ID).resolve(FsUtil.safeId(submissionId)) - .resolve(CURRENT_FILE); - } - - /* default */ Path signWorkflowRoot() { - return this.root.resolve("sign-workflows").resolve(BY_ID); - } - // ------------------------------------------------------------------------- // Revocations (single authoritative journal) // ------------------------------------------------------------------------- diff --git a/pki/src/main/java/zeroecho/pki/spi/store/MetadataKey.java b/pki/src/main/java/zeroecho/pki/spi/store/MetadataKey.java index 78333b5..c75be1a 100644 --- a/pki/src/main/java/zeroecho/pki/spi/store/MetadataKey.java +++ b/pki/src/main/java/zeroecho/pki/spi/store/MetadataKey.java @@ -1,5 +1,9 @@ package zeroecho.pki.spi.store; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Objects; @@ -16,6 +20,11 @@ import java.util.Optional; * @param key exact logical key */ public record MetadataKey(String namespace, String key) implements Comparable { + private static final String CANONICAL_VERSION_PREFIX = "mk1:"; + private static final byte[] CANONICAL_VERSION_BYTES = + CANONICAL_VERSION_PREFIX.getBytes(StandardCharsets.US_ASCII); + private static final byte LENGTH_SEPARATOR = ':'; + private static final int MAXIMUM_CANONICAL_CHARACTERS = 4364; private static final int MINIMUM_KEY_BYTE = 0x21; private static final int MAXIMUM_KEY_BYTE = 0x7e; /** Maximum canonical namespace length in UTF-8 bytes. */ @@ -33,6 +42,9 @@ public record MetadataKey(String namespace, String key) implements Comparable MAXIMUM_NAMESPACE_UTF8_BYTES) { + throw new IllegalArgumentException("Metadata namespace exceeds its canonical UTF-8 limit"); + } if (!value.matches("[a-z0-9]+(?:[a-z0-9-]*[a-z0-9])?(?:\\.[a-z0-9]+(?:[a-z0-9-]*[a-z0-9])?)*")) { throw new IllegalArgumentException("Metadata namespace is not canonical"); } @@ -44,6 +56,9 @@ public record MetadataKey(String namespace, String key) implements Comparable MAXIMUM_KEY_UTF8_BYTES) { + throw new IllegalArgumentException("Metadata key exceeds its canonical UTF-8 limit"); + } if (!emptyAllowed && value.isEmpty()) { throw new IllegalArgumentException("Metadata key must not be empty"); } @@ -52,7 +67,6 @@ public record MetadataKey(String namespace, String key) implements Comparable MAXIMUM_KEY_BYTE - || current == ':' || current == '/' || current == '\\') { throw new IllegalArgumentException("Metadata key is not canonical visible ASCII"); @@ -103,35 +117,32 @@ public record MetadataKey(String namespace, String key) implements Comparable MAXIMUM_CANONICAL_CHARACTERS) { + throw malformedRepresentation(); + } + byte[] bytes = strictEncode(encoded); + if (!hasCanonicalPrefix(bytes)) { + throw malformedRepresentation(); + } + LengthField namespaceLength = parseLength(bytes, CANONICAL_VERSION_BYTES.length, + MAXIMUM_NAMESPACE_UTF8_BYTES); + LengthField keyLength = parseLength(bytes, namespaceLength.nextOffset(), MAXIMUM_KEY_UTF8_BYTES); + int componentLength; + int expectedEnd; + try { + componentLength = Math.addExact(namespaceLength.value(), keyLength.value()); + expectedEnd = Math.addExact(keyLength.nextOffset(), componentLength); + } catch (ArithmeticException exception) { + throw new IllegalArgumentException("Metadata key representation length overflows", exception); + } + if (namespaceLength.value() == 0 || keyLength.value() == 0 || expectedEnd != bytes.length) { + throw malformedRepresentation(); + } + int namespaceStart = keyLength.nextOffset(); + int keyStart; + try { + keyStart = Math.addExact(namespaceStart, namespaceLength.value()); + } catch (ArithmeticException exception) { + throw new IllegalArgumentException("Metadata key representation length overflows", exception); + } + MetadataKey result = new MetadataKey( + strictDecode(bytes, namespaceStart, namespaceLength.value()), + strictDecode(bytes, keyStart, keyLength.value())); + if (!result.canonical().equals(encoded)) { + throw malformedRepresentation(); + } + return result; + } + + private static LengthField parseLength(byte[] bytes, int offset, int maximum) { + if (offset >= bytes.length) { + throw malformedRepresentation(); + } + int value = 0; + int cursor = offset; + while (cursor < bytes.length && bytes[cursor] != LENGTH_SEPARATOR) { + int digit = bytes[cursor] - '0'; + if (digit < 0 || digit > 9 || cursor > offset && bytes[offset] == '0') { + throw malformedRepresentation(); + } + try { + value = Math.addExact(Math.multiplyExact(value, 10), digit); + } catch (ArithmeticException exception) { + throw new IllegalArgumentException("Metadata key representation length overflows", exception); + } + if (value > maximum) { + throw malformedRepresentation(); + } + cursor++; + } + if (cursor == offset || cursor >= bytes.length) { + throw malformedRepresentation(); + } + return new LengthField(value, cursor + 1); + } + + private static boolean hasCanonicalPrefix(byte[] bytes) { + if (bytes.length < CANONICAL_VERSION_BYTES.length) { + return false; + } + for (int index = 0; index < CANONICAL_VERSION_BYTES.length; index++) { + if (bytes[index] != CANONICAL_VERSION_BYTES[index]) { + return false; + } + } + return true; + } + + private static byte[] strictEncode(String value) { + try { + ByteBuffer encoded = StandardCharsets.UTF_8.newEncoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .encode(CharBuffer.wrap(value)); + byte[] result = new byte[encoded.remaining()]; + encoded.get(result); + return result; + } catch (CharacterCodingException exception) { + throw new IllegalArgumentException("Metadata key representation is not valid UTF-8", exception); + } + } + + private static String strictDecode(byte[] bytes, int offset, int length) { + try { + return StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .decode(ByteBuffer.wrap(bytes, offset, length)) + .toString(); + } catch (CharacterCodingException exception) { + throw new IllegalArgumentException("Metadata key component is not valid UTF-8", exception); + } + } + + private static IllegalArgumentException malformedRepresentation() { + return new IllegalArgumentException("Metadata key representation is not canonical"); + } + + private record LengthField(int value, int nextOffset) { + } + } } diff --git a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemSignWorkflowStoreTest.java b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemSignWorkflowStoreTest.java index 6143ac0..e886b61 100644 --- a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemSignWorkflowStoreTest.java +++ b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemSignWorkflowStoreTest.java @@ -39,6 +39,8 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static zeroecho.pki.testkit.PkiTestRuntime.signingAuthority; +import java.io.ByteArrayInputStream; +import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; @@ -51,6 +53,7 @@ import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.OptionalLong; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -71,8 +74,13 @@ import zeroecho.pki.api.audit.AccessContext; import zeroecho.pki.api.audit.Principal; import zeroecho.pki.api.audit.Purpose; import zeroecho.pki.api.orch.SigningSubmissionId; +import zeroecho.core.io.CancellationSignal; +import zeroecho.core.io.RepeatableContent; import zeroecho.pki.impl.core.async.PkiSigningBus; import zeroecho.pki.spi.store.SignWorkflowStore; +import zeroecho.pki.spi.store.MetadataKey; +import zeroecho.pki.spi.store.MetadataSnapshot; +import zeroecho.pki.spi.store.MetadataTransaction; import zeroecho.pki.testkit.InMemorySignatureWorkflow; final class FilesystemSignWorkflowStoreTest { @@ -128,19 +136,17 @@ final class FilesystemSignWorkflowStoreTest { } String fingerprint = intent.fingerprint(); - Path recordPath = new FsPaths(root).signWorkflowPath(id); - byte[] malformed = Files.readAllBytes(recordPath); - Files.write(recordPath, Arrays.copyOf(malformed, malformed.length - 1)); + Path recordPath = root.resolve("sign-workflows/by-id/legacy/current.bin"); + Files.createDirectories(recordPath.getParent()); + Files.write(recordPath, new byte[] { 1, 2, 3 }); try (FilesystemPkiStore reopened = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults(), clock); InMemorySignatureWorkflow signer = new InMemorySignatureWorkflow(Map.of(), false)) { - RuntimeException activationFailure = assertThrows(RuntimeException.class, - () -> new PkiSigningBus(reopened, signer, root.resolve("bus.log"), signingAuthority(signer))); - assertFalse(activationFailure.toString().contains(fingerprint)); - RuntimeException attachFailure = assertThrows(RuntimeException.class, - () -> reopened.createSignIntent(intent)); - assertFalse(attachFailure.toString().contains(fingerprint)); - assertThrows(RuntimeException.class, () -> reopened.getSignRecord(id)); + assertEquals(intent.fingerprint(), reopened.getSignRecord(id).orElseThrow().fingerprint()); + try (PkiSigningBus ignored = new PkiSigningBus(reopened, signer, root.resolve("bus.log"), + signingAuthority(signer))) { + assertEquals(intent.fingerprint(), reopened.getSignRecord(id).orElseThrow().fingerprint()); + } } System.out.println("...ok"); } @@ -161,19 +167,17 @@ final class FilesystemSignWorkflowStoreTest { store.createSignIntent(current); } - Path recordPath = new FsPaths(root).signWorkflowPath(id); - byte[] unsupported = Files.readAllBytes(recordPath); - java.nio.ByteBuffer.wrap(unsupported).putInt(Integer.BYTES, 99); - Files.write(recordPath, unsupported); + Path recordPath = root.resolve("sign-workflows/by-id/legacy/current.bin"); + Files.createDirectories(recordPath.getParent()); + Files.write(recordPath, ByteBuffer.allocate(8).putInt(0x5A455352).putInt(99).array()); try (FilesystemPkiStore reopened = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults(), clock); InMemorySignatureWorkflow signer = new InMemorySignatureWorkflow(Map.of(), false)) { - RuntimeException failure = assertThrows(RuntimeException.class, - () -> new PkiSigningBus(reopened, signer, root.resolve("unsupported-version-bus.log"), - signingAuthority(signer))); - assertTrue(failure.toString().contains("Unsupported signing record version")); - assertFalse(failure.toString().contains(fingerprint)); - assertFalse(failure.toString().contains(id.value())); + assertEquals(id, reopened.getSignRecord(id).orElseThrow().submissionId()); + try (PkiSigningBus ignored = new PkiSigningBus(reopened, signer, + root.resolve("unsupported-version-bus.log"), signingAuthority(signer))) { + assertEquals(id, reopened.getSignRecord(id).orElseThrow().submissionId()); + } } System.out.println("...ok"); } @@ -256,12 +260,8 @@ final class FilesystemSignWorkflowStoreTest { store.exportSnapshot(snapshot, createdAt.minusSeconds(60)); } try (FilesystemPkiStore restored = new FilesystemPkiStore(snapshot, FsPkiStoreOptions.defaults(), clock)) { - SignWorkflowStore.Record record = restored.getSignRecord(id).orElseThrow(); assertEquals(namespace.substring(0, 32), restored.signingNamespace()); - assertEquals(SignWorkflowStore.State.SUCCEEDED, record.state()); - assertEquals(3L, record.revision()); - assertEquals(1L, record.fence()); - assertEquals(9, record.result().orElseThrow().bytes()[0]); + assertTrue(restored.getSignRecord(id).isEmpty()); assertEquals(createdAt.plusSeconds(30), restored.signingNow()); } } @@ -624,13 +624,45 @@ final class FilesystemSignWorkflowStoreTest { } private static void writeRawCurrentRecord(Path root, PkiId id, SignWorkflowStore.Record record) throws Exception { - Path path = new FsPaths(root).signWorkflowPath(id); - byte[] existing = Files.readAllBytes(path); byte[] payload = FsCodec.encode(FsCodec.SIGN_WORKFLOW_RECORD, record); - ByteBuffer replacement = ByteBuffer.allocate(Integer.BYTES * 2 + payload.length); - replacement.put(existing, 0, Integer.BYTES * 2); - replacement.put(payload); - Files.write(path, replacement.array()); + MetadataKey key = new MetadataKey("io.zeroecho.pki.signing-record", id.value()); + try (PosixTransactionalMetadataStore metadata = PosixTransactionalMetadataStore.open( + new FsPaths(root).transactionalMetadataLog()); + MetadataSnapshot snapshot = metadata.snapshot(); + MetadataTransaction transaction = metadata.beginTransaction()) { + long revision = snapshot.get(key).orElseThrow().recordRevision(); + transaction.replace(key, revision, new TestContent(payload), CancellationSignal.NONE); + assertEquals(zeroecho.pki.spi.store.MetadataCommitResult.Outcome.COMMITTED, + transaction.commit().outcome()); + } + } + + private static final class TestContent implements RepeatableContent { + private final byte[] value; + + private TestContent(byte[] value) { + this.value = value.clone(); + } + + @Override + public InputStream openStream() { + return new ByteArrayInputStream(value); + } + + @Override + public OptionalLong length() { + return OptionalLong.of(value.length); + } + + @Override + public String contentId() { + return "test-sign-record"; + } + + @Override + public void close() { + Arrays.fill(value, (byte) 0); + } } private static void assertRedactedStructuralFailure(byte[] encoded, String sensitiveFingerprint) { diff --git a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemStagedContentStoreTest.java b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemStagedContentStoreTest.java index df99d0a..408b93b 100644 --- a/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemStagedContentStoreTest.java +++ b/pki/src/test/java/zeroecho/pki/impl/fs/FilesystemStagedContentStoreTest.java @@ -46,6 +46,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.HexFormat; import java.util.List; +import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -139,6 +140,37 @@ final class FilesystemStagedContentStoreTest { System.out.println("recoveryRetainsOnlyDurablyReferencedCompletedContent...ok"); } + @Test + void transactionalSigningOwnerPreventsGeneralRetirement(@TempDir Path directory) throws Exception { + System.out.println("transactionalSigningOwnerPreventsGeneralRetirement"); + FilesystemStagedContentStore store = new FilesystemStagedContentStore(directory, STORE_ID); + DurableContentReference reference = stage(store, DurableContentReference.Lifecycle.OPERATION, "sign-input"); + DurableContentOwner authority = signingOwner("authority"); + store.bindSigningOwnership(candidate -> candidate.equals(reference) ? Set.of(authority) : Set.of()); + assertThrows(IOException.class, () -> store.retireUnownedContent(reference)); + try (zeroecho.core.io.RepeatableContent content = store.openContent(reference)) { + assertEquals(reference.length(), content.length().orElseThrow()); + } + assertThrows(IllegalArgumentException.class, + () -> store.retainContent(reference, signingOwner("authority"))); + System.out.println("transactionalSigningOwnerPreventsGeneralRetirement...ok"); + } + + @Test + void sharedSigningPublicationReservationsAreReferenceCounted(@TempDir Path directory) throws Exception { + System.out.println("sharedSigningPublicationReservationsAreReferenceCounted"); + FilesystemStagedContentStore store = new FilesystemStagedContentStore(directory, STORE_ID); + DurableContentReference reference = stage(store, DurableContentReference.Lifecycle.OPERATION, "shared"); + try (FilesystemStagedContentStore.SigningReservation first = store.reserveSigningPublication(reference); + FilesystemStagedContentStore.SigningReservation second = store.reserveSigningPublication(reference)) { + first.close(); + assertThrows(IOException.class, () -> store.retireUnownedContent(reference)); + } + store.retireUnownedContent(reference); + assertThrows(IOException.class, () -> store.openContent(reference)); + System.out.println("sharedSigningPublicationReservationsAreReferenceCounted...ok"); + } + @Test void rejectsTraversalForeignMetadataAndSymlinkContent(@TempDir Path directory) throws Exception { System.out.println("rejectsTraversalForeignMetadataAndSymlinkContent"); diff --git a/pki/src/test/java/zeroecho/pki/spi/store/MetadataKeyTest.java b/pki/src/test/java/zeroecho/pki/spi/store/MetadataKeyTest.java index f70c6a1..bf7321e 100644 --- a/pki/src/test/java/zeroecho/pki/spi/store/MetadataKeyTest.java +++ b/pki/src/test/java/zeroecho/pki/spi/store/MetadataKeyTest.java @@ -28,10 +28,11 @@ final class MetadataKeyTest { @Test void canonicalExtensionIdentityRoundTrips() { System.out.print("canonicalExtensionIdentityRoundTrips "); - MetadataKey key = MetadataKey.parse("example.audit:Record-01"); + MetadataKey key = new MetadataKey("example.audit", "zsign:v1:tenant:42:nonce"); assertEquals("example.audit", key.namespace()); - assertEquals("Record-01", key.key()); + assertEquals("zsign:v1:tenant:42:nonce", key.key()); assertEquals(key, MetadataKey.parse(key.canonical())); + assertEquals(key.canonical(), key.toString()); System.out.println("...ok"); } @@ -40,10 +41,10 @@ final class MetadataKeyTest { System.out.print("reservedAndAmbiguousIdentitiesFail "); assertThrows(IllegalArgumentException.class, () -> new MetadataKey("zeroecho", "a")); assertThrows(IllegalArgumentException.class, () -> new MetadataKey("zeroecho.audit", "a")); - assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("Example.audit:a")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("mk1:13:1:Example.audita")); assertThrows(IllegalArgumentException.class, () -> new MetadataKey("example.audit", "a/b")); assertThrows(IllegalArgumentException.class, () -> new MetadataKey("example.audit", "é")); - assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("example.audit:a:b")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("example.audit:a")); System.out.println("...ok"); } @@ -52,12 +53,16 @@ final class MetadataKeyTest { System.out.print("orderingAndPrefixRangesAreTotal "); List sorted = List.of( new MetadataKey("example.audit", "a~"), + new MetadataKey("example.audit", ":"), + new MetadataKey("example.audit", ":child"), new MetadataKey("example.audit", "abc"), new MetadataKey("example.audit", "~")) .stream() .sorted() .toList(); - assertEquals("abc", sorted.get(0).key()); + assertEquals(":", sorted.get(0).key()); + assertEquals(":child", sorted.get(1).key()); + assertEquals("abc", sorted.get(2).key()); MetadataSnapshot.KeyRange ordinary = MetadataSnapshot.KeyRange.prefix("example.audit", "abc"); assertEquals("abd", ordinary.upperExclusive().orElseThrow()); @@ -106,11 +111,25 @@ final class MetadataKeyTest { assertThrows(IllegalArgumentException.class, () -> new MetadataKey(oversizedNamespace, "key")); assertThrows(IllegalArgumentException.class, - () -> MetadataKey.parse(oversizedNamespace + ":key")); + () -> MetadataKey.parse("mk1:256:3:" + oversizedNamespace + "key")); assertThrows(IllegalArgumentException.class, () -> new MetadataKey("example.audit", oversizedKey)); assertThrows(IllegalArgumentException.class, - () -> MetadataKey.parse("example.audit:" + oversizedKey)); + () -> MetadataKey.parse("mk1:13:4097:example.audit" + oversizedKey)); + System.out.println("...ok"); + } + + @Test + void malformedLengthFramingFailsClosed() { + System.out.print("malformedLengthFramingFailsClosed "); + MetadataKey key = new MetadataKey("example.audit", "zsign:v1:a:b:c"); + String canonical = key.canonical(); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("mk2:13:1:example.audita")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("mk1:013:1:example.audita")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("mk1:13:2:example.audita")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse(canonical.substring(0, canonical.length() - 1))); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse(canonical + "x")); + assertThrows(IllegalArgumentException.class, () -> MetadataKey.parse("mk1:999999999999:1:xz")); System.out.println("...ok"); } }