dateKeys-dart/test/open_stream_test.dart

// Streaming: capsules whose content is several MiB, made from the fixtures
// (large_capsule.dart), opened from a source read in pieces, which holds
// only the bytes that steps 1 to 8 need and streams PAYLOAD_AGE, and in
// memory, with the same result. The output receives the content as age
// authenticates it and is closed only at step 18; a failure in a later
// chunk aborts it, never closed (spec §56). The capsule_digest of a .dkk is
// computed over the source.
@TestOn('vm')
library;

import 'dart:convert';
import 'dart:io';
import 'dart:typed_data';

import 'package:datekeys/datekeys.dart';
import 'package:datekeys/src/age.dart' show maxAgeHeaderLength;
import 'package:datekeys/src/sha256.dart' show sha256;
import 'package:test/test.dart';

import 'large_capsule.dart';
import 'open_support.dart';

Json readJson(String path) => jsonDecode(File(path).readAsStringSync()) as Json;

Uint8List fixture(String file) =>
    File('testdata/fixtures/$file').readAsBytesSync();

Release releaseOf(Json r) {
  final rel = r['release']! as Json;
  return Release(rel['round']! as int, fromHex(str(rel, 'signature')));
}

void main() {
  final r1 = readJson('testdata/fixtures/time_only.json');
  final r3 = readJson('testdata/fixtures/format3_single.json');
  final content = pattern((3 << 20) + 12345);
  final big1 = withContent1(
    fixture('time_only.dkc'),
    fromHex(str(r1, 'payload_identity')),
    content,
  );
  OpenOptions options1({ByteSink? output, AccessKey? key}) => OpenOptions(
    source: suppliedRelease(releaseOf(r1)),
    now: () => parseRfc3339(str(r1, 'unlock_at')),
    output: output,
    accessKey: key,
  );

  test('format 1: 3 MiB from a source, in pieces, as in memory', () async {
    final source = ChunkySource(big1, 1 << 20);
    final out = RecordingOutput();
    final r = await openCapsuleSource(source, options1(output: out));
    expect(r.ok, isTrue);
    expect(out.received.toBytes(), content);
    expect([r.payloadLength, out.closed], [content.length, true]);
    // Each chunk of 64 KiB as age authenticates it, and the close last.
    expect(out.log.where((l) => l == 'add 65536'), hasLength(48));
    expect(out.log.last, 'close');
    // The source was read once in prefix, and then in pieces of at most
    // 1 MiB, never whole.
    final payloadAt = inspectCapsule(big1).payloadOffset!;
    expect(source.largest, lessThanOrEqualTo(1 << 20));
    expect(source.farthest, big1.length);
    expect(
      source.bytesRead,
      lessThan(big1.length + payloadAt + maxAgeHeaderLength + 100),
    );
    final memory = RecordingOutput();
    final m = await openCapsule(big1, options1(output: memory));
    expect([m.ok, memory.received.toBytes()], [true, content]);
    expect(
      canonical([for (final c in m.checks) c.toJson()]),
      canonical([for (final c in r.checks) c.toJson()]),
    );
  });

  test('format 1: a chunk that fails after others were written aborts the '
      'output, never closed', () async {
    for (final at in [
      big1.length - 10,
      big1.length - (1 << 20),
      big1.length ~/ 2,
    ]) {
      final bad = Uint8List.fromList(big1)..[at] ^= 1;
      for (final fromSource in [false, true]) {
        final out = RecordingOutput();
        final r = fromSource
            ? await openCapsuleSource(
                ChunkySource(bad, 1 << 20),
                options1(output: out),
              )
            : await openCapsule(bad, options1(output: out));
        expect(
          r.error!.message,
          'capsule: PAYLOAD_AGE: the age payload is '
          'truncated, has trailing data or fails STREAM authentication: '
          'ERR_INTEGRITY',
        );
        expect(r.checks.last.step, 17);
        expect(out.log.where((l) => l.startsWith('add ')), isNotEmpty);
        expect([out.closed, out.aborted], [false, same(r.error)]);
        expect(out.log.last, 'abort');
      }
    }
  });

  test('format 1: the capsule_digest of a .dkk over a large source', () async {
    // The .dkk of time_and_key_portable, for a capsule that is not its own:
    // its capsule_digest fails at step 9, read from the source; one whose
    // digest is that of the capsule passes it.
    final rk = readJson('testdata/fixtures/time_and_key_portable.json');
    final big = withContent1(
      fixture('time_and_key_portable.dkc'),
      fromHex(str(rk, 'payload_identity')),
      content,
    );
    final key = decodeAccessKey(fixture('time_and_key_portable.dkk'));
    OpenOptions options(AccessKey k, ByteSink out) => OpenOptions(
      source: suppliedRelease(releaseOf(rk)),
      now: () => parseRfc3339(str(rk, 'unlock_at')),
      accessKey: k,
      output: out,
    );
    final source = ChunkySource(big, 1 << 20);
    final r = await openCapsuleSource(source, options(key, RecordingOutput()));
    expect(
      [r.error?.message, r.checks.last.step],
      [
        'capsule: the .dkk capsule_digest does not match this .dkc: '
            'ERR_ACCESS_INVALID',
        9,
      ],
    );
    expect(source.farthest, big.length);
    final own = AccessKey(
      credentialId: key.credentialId,
      capsuleId: key.capsuleId,
      type: key.type,
      material: key.material,
      verification: Verification(sha256(big)),
    );
    final out = RecordingOutput();
    final s = await openCapsuleSource(
      ChunkySource(big, 1 << 20),
      options(own, out),
    );
    expect([s.ok, out.received.toBytes()], [true, content]);
  });

  final files = [
    ('a.txt', pattern(1000)),
    ('fotos/grande.bin', pattern((3 << 20) + 7)),
    ('vacio', Uint8List(0)),
  ];
  final big3 = withFiles3(
    fixture('format3_single.dkc'),
    fromHex(str(r3, 'payload_identity')),
    releaseOf(r3),
    files,
  );
  OpenOptions options3(FileSink sink) => OpenOptions(
    source: suppliedRelease(releaseOf(r3)),
    now: () => parseRfc3339(str(r3, 'unlock_at')),
    sink: sink,
  );

  test('format 3: files of 3 MiB from a source, in pieces, as in '
      'memory', () async {
    for (final fromSource in [false, true]) {
      final sink = RecordingSink();
      final r = fromSource
          ? await openCapsuleSource(ChunkySource(big3, 1 << 20), options3(sink))
          : await openCapsule(big3, options3(sink));
      expect(r.ok, isTrue, reason: r.error?.message);
      expect(
        [for (final f in r.head!.files) f.path],
        [for (final (p, _) in files) p],
      );
      for (var i = 0; i < files.length; i++) {
        expect(sink.files[i].toBytes(), files[i].$2);
      }
      expect(sink.log.last, 'commit');
      expect(sink.log.where((l) => l == 'abort'), isEmpty);
    }
  });

  test('format 3: a chunk that fails after a file was written aborts the '
      'sink', () async {
    final bad = Uint8List.fromList(big3)..[big3.length - 100] ^= 1;
    for (final fromSource in [false, true]) {
      final sink = RecordingSink();
      final r = fromSource
          ? await openCapsuleSource(ChunkySource(bad, 1 << 20), options3(sink))
          : await openCapsule(bad, options3(sink));
      expect(
        r.error!.message,
        'capsule: PAYLOAD_AGE: the age payload is '
        'truncated, has trailing data or fails STREAM authentication: '
        'ERR_INTEGRITY',
      );
      expect(sink.log, contains('close 0'));
      expect(
        [sink.committed, sink.aborted, sink.log.last],
        [false, same(r.error), 'abort'],
      );
    }
  });
}