decryptStream method

Stream<Uint8List> decryptStream({ required Uint8List recipientSecretKey, required File input, Uint8List? recipientKexSecretKey, String? recipientKeyId, Uint8List? signerPublicKey, Uint8List? aadResolver(PqStreamingHeader header)?, void onHeader(PqStreamingHeader header)?, })

Streams the authenticated plaintext of a .pqfs input, one frame at a time, without materializing it anywhere — the consumer decides what to do with each frame (write it, parse it, forward it).

Every frame is verified (tag + position + finality) before it is yielded, so a consumer only ever sees authentic bytes; truncation is detected at the end of the stream. onHeader fires once after the header (and its signature, when present) has been verified.

Hybrid containers (those whose header metadata carries the hybridKex marker) require recipientKexSecretKey — the raw 32-byte X25519 secret matching the public key the sender encrypted to; a missing key fails before any frame is read unless the container also carries recipients[] entries, which are then still tried. The key is ignored for non-hybrid containers.

Multi-recipient containers resolve the DEM key on the first frame: the primary derivation and the recipients[] unwrap are trial-opened and the authenticating one is kept. recipientKeyId (this key's id) routes straight to the matching wrap entry. A non-default AEAD suite recorded in the header (aeadSuite) rebuilds the engine on the same provider.

Implementation

Stream<Uint8List> decryptStream({
  required Uint8List recipientSecretKey,
  required File input,
  Uint8List? recipientKexSecretKey,
  String? recipientKeyId,
  Uint8List? signerPublicKey,
  Uint8List? Function(PqStreamingHeader header)? aadResolver,
  void Function(PqStreamingHeader header)? onHeader,
}) async* {
  final source = await input.open();
  try {
    final container = await _readContainerHeader(source);
    final header = container.header;

    if (container.signature != null) {
      if (signerPublicKey == null) {
        throw const PqForgeException(
          'signerPublicKey is required for a signed streaming envelope',
        );
      }
      final ok = PqSignaturePrimitives.verify(
        header.signatureAlgorithm!,
        signerPublicKey,
        container.headerCore,
        container.signature!,
        context: PqStreamingEnvelope.signatureContext(),
        preHash: true,
      );
      if (!ok) {
        throw const PqForgeException(
          'ML-DSA streaming header signature verification failed',
        );
      }
    }

    final aad = aadResolver?.call(header);
    if (header.aadHash != null) {
      if (aad == null) {
        throw const PqForgeException(
          'AAD is required for this streaming envelope',
        );
      }
      if (!PqBytes.constantTimeEquals(PqBytes.sha256(aad), header.aadHash!)) {
        throw const PqForgeException('AAD hash mismatch');
      }
    }
    onHeader?.call(header);

    // The container records a non-default AEAD suite in its header
    // metadata; rebuild the engine on the same provider to match it.
    final suite = PqAeadSuite.of(header.metadata);
    PqFipsMode.requireApprovedSuite(suite);
    final aead = engine.cipherSuite == suite
        ? engine
        : aeadEngineForProvider(engine.provider, cipherSuite: suite);

    final sharedSecret = PqKemPrimitives.decapsulate(
      header.kemAlgorithm,
      recipientSecretKey,
      header.kemCiphertext,
    );
    final hasEntries = PqMultiRecipient.hasEntries(header.metadata);

    // Primary derivation. On a multi-recipient container its failure (e.g.
    // a hybrid primary but no kex key — we may not BE the primary) is
    // deferred so the recipients[] entries still get their chance.
    Uint8List? primaryDemKey;
    PqForgeException? primaryFailure;
    try {
      if (PqHybridKemDem.isHybrid(header.metadata)) {
        if (recipientKexSecretKey == null) {
          throw const PqForgeException(
            'This container uses hybrid ML-KEM + X25519 encryption; the '
            'recipient X25519 secret key is required to decrypt it',
          );
        }
        primaryDemKey = await PqHybridKemDem.demKeyForOpen(
          profile: header.profile,
          kemSharedSecret: sharedSecret,
          kemCiphertext: header.kemCiphertext,
          metadata: header.metadata,
          recipientKexSecretKey: recipientKexSecretKey,
        );
      } else {
        primaryDemKey = PqForge.deriveDemKey(
          header.profile,
          sharedSecret,
          header.kemCiphertext,
        );
      }
    } on PqForgeException catch (failure) {
      if (!hasEntries) rethrow;
      primaryFailure = failure;
    }

    // Candidate DEM keys, primary first unless the metadata names a
    // different primary key id than ours. With several candidates the one
    // that authenticates the first frame wins and is kept for the rest.
    final candidates = <Uint8List>[];
    if (!hasEntries) {
      candidates.add(primaryDemKey!);
    } else {
      final entriesKey = await PqMultiRecipient.unwrapDemKey(
        profile: header.profile,
        metadata: header.metadata,
        recipientSecretKey: recipientSecretKey,
        recipientKexSecretKey: recipientKexSecretKey,
        recipientKeyId: recipientKeyId,
      );
      final namedPrimary = header.metadata[PqMultiRecipient.primaryKeyIdKey];
      final preferEntries =
          recipientKeyId != null &&
          namedPrimary is String &&
          namedPrimary != recipientKeyId;
      for (final key
          in preferEntries
              ? [entriesKey, primaryDemKey]
              : [primaryDemKey, entriesKey]) {
        if (key != null) candidates.add(key);
      }
      if (candidates.isEmpty) {
        throw PqForgeException(
          'This multi-recipient container is not addressed to the supplied '
          'key: the primary derivation was unavailable '
          '(${primaryFailure?.message}) and no recipients[] entry unwrapped',
        );
      }
    }
    // Multi-recipient containers always resolve via the first-frame trial,
    // even with a single candidate, so a wrong key surfaces as the
    // descriptive "not addressed" error rather than a bare tag failure.
    var demKey = hasEntries ? null : candidates.single;

    final tagLength = aead.cipherSuite.tagLength;
    final maxBody = header.frameSize + tagLength;

    // One frame of read-ahead (R9): the next frame loads from disk while
    // the current one is opened. The observable validation order is
    // unchanged — a prefetched frame is only inspected after every earlier
    // frame has authenticated and been yielded.
    Future<({int seq, bool isFinal, Uint8List body})?> readFrame() async {
      final frameHeader = await _readExactlyOrNull(
        source,
        PqStreamingEnvelope.frameHeaderBytes,
      );
      if (frameHeader == null) return null; // clean EOF at a frame boundary
      final parsed = PqStreamingEnvelope.parseFrameHeader(frameHeader);
      // Bound the body length BEFORE allocating it (anti-DoS).
      if (parsed.bodyLen < tagLength || parsed.bodyLen > maxBody) {
        throw PqForgeException(
          'Invalid streaming frame body length: ${parsed.bodyLen}',
        );
      }
      return (
        seq: parsed.seq,
        isFinal: parsed.isFinal,
        body: await _readExactly(source, parsed.bodyLen),
      );
    }

    var expectedSeq = 0;
    var sawFinal = false;
    var current = await readFrame();
    while (current != null) {
      if (sawFinal) {
        throw const PqForgeException(
          'Trailing data after the final streaming frame',
        );
      }
      if (current.seq != expectedSeq) {
        throw PqForgeException(
          'Streaming frame out of order: expected $expectedSeq, '
          'got ${current.seq}',
        );
      }
      final pending = current.isFinal ? null : readFrame();
      Uint8List plaintext;
      try {
        if (demKey != null) {
          plaintext = await PqStreamingEnvelope.openFrameBody(
            engine: aead,
            demKey: demKey,
            headerHash: container.headerHash,
            nonceSalt: header.nonceSalt,
            seq: expectedSeq,
            isFinal: current.isFinal,
            body: current.body,
          );
        } else {
          // First frame of a multi-recipient container: trial-open the
          // candidates and keep the one that authenticates.
          Uint8List? opened;
          for (final candidate in candidates) {
            try {
              opened = await PqStreamingEnvelope.openFrameBody(
                engine: aead,
                demKey: candidate,
                headerHash: container.headerHash,
                nonceSalt: header.nonceSalt,
                seq: expectedSeq,
                isFinal: current.isFinal,
                body: current.body,
              );
              demKey = candidate;
              break;
            } on PqForgeAuthTagException {
              continue;
            }
          }
          plaintext =
              opened ??
              (throw const PqForgeException(
                'This multi-recipient container is not addressed to the '
                'supplied key: no candidate DEM key authenticated',
              ));
        }
      } on Object {
        // Never leave the prefetch dangling: closing the source in the
        // finally below would fail it into an unawaited future.
        await _drainQuietly(pending);
        rethrow;
      }
      if (current.isFinal) sawFinal = true;
      expectedSeq++;
      yield plaintext;
      current = pending == null ? await readFrame() : await pending;
    }
    if (!sawFinal) {
      throw const PqForgeException(
        'Truncated streaming envelope: missing final frame',
      );
    }
  } finally {
    await source.close();
  }
}