Preserve and verify SBI companions during disc setup

This commit is contained in:
Alexandros Mandravillis
2026-09-05 20:51:14 +03:00
parent 17f49ad3b2
commit 1c60d340dc
7 changed files with 464 additions and 5 deletions
+92
View File
@@ -0,0 +1,92 @@
# Disc subchannel companions
An SBI companion contains disc subchannel metadata that a main-track image can omit.
`tools/prepare_disc.py` retains a user-supplied companion when it copies or converts a disc.
`psxrecomp_cli.py verify-disc` and `generate` check it before generation.
These tools do not supply, download, or license SBI files.
## Supply a companion
1. Obtain the matching SBI file through a lawful source.
2. Put it beside the input image with the same basename and the `.sbi` extension.
3. For CUE input, use the CUE basename, even when its first BIN has a different name.
4. Run preparation or verification again.
For example, `Album.cue` uses `Album.sbi`.
Preparation accepts `.SBI` too, then writes the output CUE basename with a lowercase `.sbi` extension.
It retains the companion bytes unchanged, including when the output CUE has a new name.
The runtime loads that companion beside the mounted CUE or image.
An SBI beside the first track does not substitute for one beside the selected CUE.
Preparation refuses conflicting output companions before it writes disc files.
Use an empty output directory or move the conflicting companion before retrying.
An identical existing companion remains untouched.
Concurrent preparation into the same directory is unsupported.
## Exact revision coverage
The registry in `tools/disc_companion.py` requires both measured main-track size and SHA-1.
It does not infer protection from a region, filename, or configured serial.
The initial supported identity is:
| Field | Measured identity |
|---|---|
| Title / serial | Resident Evil 3: Nemesis / SLES-02529 |
| Revision key | Europe; exact data track below, no claim for other pressings |
| Main-track size | `712491360` bytes |
| Main-track SHA-1 | `b3ec6631e1c9ed3b0f554b3a5f61166bac8ebb9d` |
| Qualified SBI SHA-256 | `ada8877a2a964eff2743d53cc043be0b7b148469b60de82fd1069f32700670eb` |
For that identity, a missing or different SBI stops setup with an actionable error.
`--skip-hash-check` does not bypass the companion gate.
The matching companion must also pass the supported nonempty SBI type-1 record checks.
A different valid SBI representation needs maintainer qualification before this gate accepts it.
Unknown identities remain `unknown` when no SBI exists.
A supplied, structurally valid SBI for an unknown identity is `format_valid_unqualified`.
Neither result proves complete subchannels or compatibility with that revision.
Cooked ISO and 2448-byte conversions preserve supplied SBI files but do not inherit the raw-track registry identity.
The existing 2448-byte conversion discards embedded subchannels; this change does not extract them into SBI.
CHD preparation and automatic protection detection for other revisions are outside this tool's coverage.
## Receipts and callers
Preparation writes `<output-cue-basename>.disc-receipt.json` beside the output.
It records the source image, measured data-track hashes, output CUE, and separate `subchannel` identity.
The companion record contains its source path, byte size, SHA-256, record count, status, and output path.
For a qualified revision, it also contains the registry identity and evidence reference.
Keep this receipt private with the user's inputs because it includes local paths.
The existing `RESULT_CUE=` output remains unchanged.
CLI disc results add the same `subchannel` record.
The legacy `verified` field describes only the existing main-track check; it never proves complete subchannels.
Missing required companions return preparation exit code 1 or CLI verification exit code 3.
No configuration keys or graphical file selectors changed.
Existing setup hosts need the updated Python tools, including `tools/disc_companion.py`, to receive this gate.
Direct runtime launches and older installed setup kits retain their existing behavior.
## Provenance and validation
The registry metadata comes from the private 2026-09-05 SLES-02529 same-executable control and operator acceptance.
The unchanged Wave 3 source was `63446a28111a21a0cb7e8b5106c3614c898fc4b5`.
The accepted executable SHA-256 was `131e8ecbd18a425ba2b5b49a30133fc0afed888af13666b653b3b5e927f3b0fe`.
It closed normally at frame 14841 with `sdl_window_close` after the operator reported correct behavior.
The same main track without SBI stalled at the warning screen.
The private portfolio report and operator receipt retain the detailed evidence; no retail or SBI payload accompanies this source.
That technical acceptance does not grant redistribution rights for the companion.
Primary format reference: [PSX-SPX CDROM format](https://psx-spx.consoledev.net/cdromformat/#cdrom-protection-libcrypt).
The existing runtime reader supplies the type-1 parsing contract in `ISOReader::LoadSBICompanion`.
Runtime support came from `cebe738fd17cc4c37a76729462e093ff16ee457a`.
Upstream `17f49ad3b20dc30917a881a02baaa25374c13d18` already contains that commit, but lacks the setup gate and copy step.
This change starts from that upstream identity and changes no runtime code.
Run the source-owned setup tests from the framework root:
```sh
python tools/tests/test_disc_companion.py -v
```
The fixtures create synthetic ISO directory records, an inert executable header, audio bytes, and SBI records.
Tests bind their own synthetic revision through an in-memory registry entry.
No fixture copies retail game data, BIOS data, or a third-party SBI.
+4
View File
@@ -83,6 +83,10 @@ python psxrecomp/psxrecomp_cli.py pgo-train \
[--disc …] [--train-secs 120] [--train-runs 3] [--json-progress]
```
Disc verification and preparation also check and retain user-supplied
[SBI companions](DISC_COMPANIONS.md). Main-track hashes alone do not prove
complete subchannel input.
`generate` normalizes the dump via `tools/prepare_disc.py` when needed, then
runs `psxrecomp-game --config game.toml` into `[recompiler] out_dir`.
+5
View File
@@ -60,6 +60,11 @@ How the two configs relate:
[audit]
```
`[prepare_disc]` digests identify the main track only. See
[disc companions](DISC_COMPANIONS.md) for the separate SBI input gate,
exact revision coverage, and preparation receipts. No SBI configuration key
is required.
## Netplay disc mount (`[netplay]`)
Optional. Online play needs the same CD geometry on every peer — data-track
+8 -1
View File
@@ -28,6 +28,7 @@ from typing import Any, Optional
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / "tools"))
from sdk_progress import ProgressReporter # noqa: E402
from disc_companion import CompanionError, inspect_companion # noqa: E402
from toolchain_pack import ( # noqa: E402
ensure_toolchain as _ensure_toolchain_pack,
resolve_toolchain_bin,
@@ -845,12 +846,17 @@ def verify_disc_path(
if path.suffix.lower() == ".cue":
path = resolve_cue_bin(path)
md5, sha1, size = file_hashes(path)
try:
subchannel, _ = inspect_companion(disc, size, sha1)
except CompanionError as exc:
raise DiscVerifyError(str(exc)) from exc
identity = {
"path": str(path),
"md5": md5,
"sha1": sha1,
"size": size,
"verified": False,
"subchannel": subchannel,
}
progress.event("disc", **identity)
sizes = [int(s) for s in (prep.get("known_sizes") or [])]
@@ -906,7 +912,8 @@ def cmd_verify_disc(args: argparse.Namespace, progress: ProgressReporter) -> int
except DiscVerifyError as exc:
progress.error(str(exc), code=EXIT_VERIFY, verify_failed=True)
return EXIT_VERIFY
progress.phase("done", pct=1.0, message="Disc OK")
progress.phase("done", pct=1.0,
message=f"Main track accepted; subchannel status: {identity['subchannel']['status']}")
progress.result(ok=True, **identity)
return EXIT_OK
+119
View File
@@ -0,0 +1,119 @@
"""Exact-revision SBI intake. No media payloads or network access.
The registry contains measured identities, not a region-based protection list.
See docs/DISC_COMPANIONS.md for provenance and coverage limits.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
class CompanionError(ValueError):
pass
# Key: (main-track byte size, SHA-1). Values are metadata, never SBI records.
REQUIRED_SBI = {
(712491360, "b3ec6631e1c9ed3b0f554b3a5f61166bac8ebb9d"): {
"title": "Resident Evil 3: Nemesis",
"serial": "SLES-02529",
"revision": "Europe, data-track SHA-1 b3ec6631e1c9ed3b0f554b3a5f61166bac8ebb9d",
"sbi_sha256": "ada8877a2a964eff2743d53cc043be0b7b148469b60de82fd1069f32700670eb",
"evidence": "RE3 same-executable operator acceptance, 2026-09-05; docs/DISC_COMPANIONS.md",
},
}
def validate_sbi(data: bytes) -> int:
"""Check the nonempty type-1 format accepted by ISOReader::LoadSBICompanion."""
if data[:4] != b"SBI\0" or len(data) < 18 or (len(data) - 4) % 14:
raise CompanionError("Invalid SBI: expected a header and complete type-1 records.")
frames: set[int] = set()
for offset in range(4, len(data), 14):
record = data[offset:offset + 14]
if record[3] != 1 or any((b & 15) > 9 or (b >> 4) > 9 for b in record[:3]):
raise CompanionError("Invalid SBI: unsupported record type or invalid BCD position.")
mm, ss, ff = [(b >> 4) * 10 + (b & 15) for b in record[:3]]
frame = mm * 4500 + ss * 75 + ff
if ss >= 60 or ff >= 75 or frame < 150 or frame in frames:
raise CompanionError("Invalid SBI: out-of-range or duplicate disc position.")
frames.add(frame)
return len(frames)
def inspect_companion(image: Path, size: int, sha1: str) -> tuple[dict, bytes | None]:
"""Inspect only the selected image's companion; never guess from another file.
CUE input uses the CUE basename, matching the runtime mount contract.
Case-insensitive discovery permits .SBI on case-sensitive source volumes.
"""
image = image.resolve()
expected = image.with_suffix(".sbi")
requirement = REQUIRED_SBI.get((size, sha1.lower()))
matches = [p for p in image.parent.iterdir() if p.name.casefold() == expected.name.casefold()]
if len(matches) > 1:
raise CompanionError(f"Ambiguous SBI companions for {image}: {matches}. Keep one matching file.")
report = {
"status": "missing" if requirement else "unknown",
"required": bool(requirement),
"revision": requirement,
"expected_path": str(expected),
}
if not matches:
if requirement:
raise CompanionError(
f"Missing SBI for {requirement['title']} ({requirement['serial']}, "
f"{requirement['revision']}). Supply your lawfully obtained matching file at "
f"{expected}, then retry. A matching main-track hash does not include subchannel data. "
"Setup does not supply or download SBI files."
)
return report, None
companion = matches[0]
try:
# A type-1 SBI cannot exceed one record per addressable BCD disc frame.
if companion.stat().st_size > 63000004:
raise CompanionError(f"SBI exceeds the supported size: {companion}")
data = companion.read_bytes()
except OSError as exc:
raise CompanionError(f"Cannot read SBI companion {companion}: {exc}") from exc
count = validate_sbi(data)
digest = hashlib.sha256(data).hexdigest()
if requirement and digest != requirement["sbi_sha256"]:
raise CompanionError(
f"SBI identity is not qualified for {requirement['serial']} ({requirement['revision']}): "
f"{companion}. Expected SHA-256 {requirement['sbi_sha256']}; got {digest}. "
"Supply the matching lawful companion or ask the port maintainer to qualify this variant."
)
report.update(status="verified" if requirement else "format_valid_unqualified",
path=str(companion), sha256=digest, size=len(data), records=count)
return report, data
def check_destination(cue: Path, data: bytes | None) -> None:
"""Refuse stale or conflicting companions before preparation writes media."""
expected = cue.with_suffix(".sbi")
if not expected.parent.exists():
return
matches = [p for p in expected.parent.iterdir() if p.name.casefold() == expected.name.casefold()]
if len(matches) > 1:
raise CompanionError(f"Ambiguous output SBI companions for {cue}. Keep only {expected}.")
for path in matches:
if data is None or not path.is_file() or path.read_bytes() != data:
raise CompanionError(f"Conflicting output SBI: {path}. Use an empty output directory or move this file, then retry.")
if path.name != expected.name and not expected.exists():
raise CompanionError(f"Output SBI uses a different filename case: {path}. Rename it to {expected}, then retry.")
def stage_companion(cue: Path, report: dict, data: bytes | None) -> dict:
check_destination(cue, data)
result = dict(report)
if data is not None:
destination = cue.with_suffix(".sbi")
if not destination.exists():
# Exclusive creation protects a companion installed since preflight.
with destination.open("xb") as stream:
stream.write(data)
result["output_path"] = str(destination.resolve())
return result
+30 -4
View File
@@ -23,6 +23,7 @@ from __future__ import annotations
import argparse
import hashlib
import json
import re
import shutil
import struct
@@ -30,6 +31,8 @@ import sys
from dataclasses import dataclass, field
from pathlib import Path
from disc_companion import CompanionError, check_destination, inspect_companion, stage_companion
DST_SEC = 2352
SRC_2448 = 2448
USER = 2048
@@ -549,8 +552,7 @@ def main() -> int:
print(f"source not found: {src}", file=sys.stderr)
return 1
cfg.out_dir.mkdir(parents=True, exist_ok=True)
selected_image = src
cue_src: Path | None = None
cue_bins: list[Path] = []
if src.suffix.lower() == ".cue":
@@ -570,6 +572,14 @@ def main() -> int:
kind = detect_kind(src, src_size)
# Bind the selected CUE basename, not its first track's basename.
try:
companion, companion_data = inspect_companion(selected_image, src_size, src_sha1)
check_destination(cfg.out_dir / cfg.cue_name, companion_data)
except CompanionError as exc:
print(str(exc), file=sys.stderr)
return 1
known_hit = matches_known(cfg, src_size, src_md5, src_sha1)
if cfg.known and not cfg.skip_hash_check:
if known_hit:
@@ -589,6 +599,22 @@ def main() -> int:
elif not cfg.known and not cfg.skip_hash_check and kind == "bin2352":
print(" no prepare_disc.known_* configured - verifying boot EXE only")
cfg.out_dir.mkdir(parents=True, exist_ok=True)
def finish(cue_path: Path) -> None:
report = stage_companion(cue_path, companion, companion_data)
receipt = {
"schema": "psxrecomp-disc-preparation-v1",
"source_image": str(selected_image),
"source_data_track": {"path": str(src), "size": src_size, "sha1": src_sha1, "md5": src_md5},
"output_cue": str(cue_path.resolve()),
"subchannel": report,
}
cue_path.with_suffix(".disc-receipt.json").write_text(
json.dumps(receipt, indent=2) + "\n", encoding="utf-8")
print(f"subchannel: {report['status']}")
print(f"RESULT_CUE={cue_path.resolve()}")
# Multi-track Redump: keep the cue + every track bin so CDDA works.
if cue_src is not None and len(cue_bins) > 1 and kind == "bin2352":
print(f" preserving multi-track Redump set ({len(cue_bins)} files)")
@@ -609,7 +635,7 @@ def main() -> int:
print(f" size {out_size}")
print(f" md5 {out_md5}")
print(f" sha1 {out_sha1}")
print(f"RESULT_CUE={cue_path}")
finish(cue_path)
return 0
if kind == "bin2352":
@@ -664,7 +690,7 @@ def main() -> int:
elif kind == "raw2448":
print(" trimmed from 2448-byte/sector dump")
print(f"RESULT_CUE={cue_path.resolve()}")
finish(cue_path)
return 0
+206
View File
@@ -0,0 +1,206 @@
"""Source-owned disc/SBI fixtures. No retail sectors, executables or SBI records."""
import contextlib
import argparse
import hashlib
import io
import json
from pathlib import Path
import struct
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[2]
sys.path[:0] = [str(ROOT), str(ROOT / "tools")]
import disc_companion as dc
import prepare_disc as prepare
import psxrecomp_cli as cli
from sdk_progress import ProgressReporter
SBI = b"SBI\0" + bytes([0, 2, 0, 1]) + bytes(range(10))
BOOT = "SLES_000.00"
def cooked_disc():
sectors = [bytearray(2048) for _ in range(24)]
sectors[16][1:6] = b"CD001"
struct.pack_into("<I", sectors[16], 158, 20)
struct.pack_into("<I", sectors[16], 166, 2048)
cursor = 0
for name, lba, data in [("SYSTEM.CNF", 21, b"BOOT = cdrom:\\" + BOOT.encode() + b";1"),
(BOOT, 22, b"PS-X EXE" + bytes(2040))]:
encoded = (name + ";1").encode()
record = bytearray(33 + len(encoded) + (len(encoded) % 2 == 0))
record[0] = len(record)
struct.pack_into("<I", record, 2, lba)
struct.pack_into("<I", record, 10, len(data))
record[32] = len(encoded)
record[33:33 + len(encoded)] = encoded
sectors[20][cursor:cursor + len(record)] = record
cursor += len(record)
sectors[lba][:len(data)] = data
return b"".join(sectors)
class CompanionTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.image = self.root / "input track.bin"
self.image.write_bytes(prepare.iso_to_bin(cooked_disc()))
self.sha1 = hashlib.sha1(self.image.read_bytes()).hexdigest()
self.size = self.image.stat().st_size
self.rule = {"title": "Source fixture", "serial": "TEST-00000", "revision": "synthetic-v1",
"sbi_sha256": hashlib.sha256(SBI).hexdigest(), "evidence": "source-owned test"}
self.registry = patch.dict(dc.REQUIRED_SBI, {(self.size, self.sha1): self.rule})
self.registry.start()
self.addCleanup(self.registry.stop)
self.config = self.root / "game.toml"
self.config.write_text(f'[game]\nid = "SLES-02529"\n[prepare_disc]\nboot_exe = "{BOOT}"\n', encoding="utf-8")
def prepare(self, source=None, out=None):
output = io.StringIO()
args = ["prepare_disc.py", str(source or self.image), "--config", str(self.config),
"--out-dir", str(out or self.root / "output"), "--cue-name", "renamed.cue"]
with patch.object(sys, "argv", args), contextlib.redirect_stdout(output), contextlib.redirect_stderr(output), patch.object(prepare, "_configure_stdio"):
code = prepare.main()
return code, output.getvalue()
def test_missing_blocks_before_output_even_when_main_hash_is_skipped(self):
code, message = self.prepare()
self.assertEqual(code, 1)
self.assertIn("Missing SBI", message)
self.assertIn(str(self.image.with_suffix(".sbi")), message)
self.assertFalse((self.root / "output").exists())
with self.assertRaisesRegex(cli.DiscVerifyError, "Missing SBI"):
cli.verify_disc_path(self.image, {}, skip_hash=True, progress=ProgressReporter())
def test_exact_registry_uses_measured_identity(self):
real_size, real_hash = next(iter(dc.REQUIRED_SBI))
with self.assertRaisesRegex(dc.CompanionError, "SLES-02529"):
dc.inspect_companion(self.image, real_size, real_hash)
report, data = dc.inspect_companion(self.image, real_size + 1, real_hash)
self.assertFalse(report["required"])
self.assertEqual(report["status"], "unknown")
self.assertIsNone(data)
def test_cli_missing_error_is_actionable_json_and_exit_three(self):
output = io.StringIO()
args = argparse.Namespace(config=str(self.config), project_root="", disc=str(self.image), skip_hash_check=True)
reporter = ProgressReporter(json_progress=True, stream=output)
code = cli.cmd_verify_disc(args, reporter)
self.assertEqual(code, 3)
error = json.loads(output.getvalue().splitlines()[-1])
self.assertEqual(error["event"], "error")
self.assertTrue(error["verify_failed"])
self.assertIn("Missing SBI", error["message"])
self.assertNotIn('"ok":true', output.getvalue())
def test_in_place_companion_is_not_rewritten(self):
cue = self.root / "renamed.cue"
cue.write_text(f'FILE "{self.image.name}" BINARY\n TRACK 01 MODE2/2352\n INDEX 01 00:00:00\n')
sbi = cue.with_suffix(".sbi")
sbi.write_bytes(SBI)
stamp = sbi.stat().st_mtime_ns
code, message = self.prepare(cue, self.root)
self.assertEqual(code, 0, message)
self.assertEqual(sbi.stat().st_mtime_ns, stamp)
self.assertEqual(sbi.read_bytes(), SBI)
def test_copy_rename_receipt_and_repeat(self):
self.image.with_suffix(".SBI").write_bytes(SBI)
before = self.image.read_bytes()
for _ in range(2):
code, message = self.prepare()
self.assertEqual(code, 0, message)
output = self.root / "output"
self.assertEqual((output / "renamed.sbi").read_bytes(), SBI)
receipt = json.loads((output / "renamed.disc-receipt.json").read_text())
self.assertEqual(receipt["subchannel"]["status"], "verified")
self.assertEqual(receipt["subchannel"]["sha256"], self.rule["sbi_sha256"])
self.assertEqual(receipt["source_data_track"]["sha1"], self.sha1)
self.assertEqual(before, self.image.read_bytes())
def test_cue_basename_multitrack_and_cdda_preserved(self):
cue = self.root / "Album.cue"
audio = self.root / "audio.bin"
audio.write_bytes(bytes([42]) * 2352)
cue.write_text(f'FILE "{self.image.name}" BINARY\n TRACK 01 MODE2/2352\n INDEX 01 00:00:00\nFILE "audio.bin" BINARY\n TRACK 02 AUDIO\n INDEX 01 00:00:00\n')
self.image.with_suffix(".sbi").write_bytes(SBI)
code, message = self.prepare(cue)
self.assertEqual(code, 1, "track companion must not substitute for CUE companion")
cue.with_suffix(".sbi").write_bytes(SBI)
code, message = self.prepare(cue)
self.assertEqual(code, 0, message)
output = self.root / "output"
self.assertEqual((output / "audio.bin").read_bytes(), audio.read_bytes())
self.assertEqual((output / "renamed.sbi").read_bytes(), SBI)
identity = cli.verify_disc_path(cue, {}, skip_hash=False, progress=ProgressReporter())
self.assertEqual(identity["subchannel"]["status"], "verified")
def test_wrong_companion_and_stale_destination_are_not_overwritten(self):
wrong = SBI[:-1] + b"\xff"
self.image.with_suffix(".sbi").write_bytes(wrong)
code, message = self.prepare()
self.assertEqual(code, 1)
self.assertIn("not qualified", message)
self.image.with_suffix(".sbi").write_bytes(SBI)
output = self.root / "output"
output.mkdir()
stale = output / "renamed.sbi"
stale.write_bytes(wrong)
code, message = self.prepare()
self.assertEqual(code, 1)
self.assertEqual(list(output.iterdir()), [stale])
self.assertEqual(stale.read_bytes(), wrong)
def test_optional_stale_companion_is_rejected(self):
with patch.dict(dc.REQUIRED_SBI, {}, clear=True):
output = self.root / "output"
output.mkdir()
(output / "renamed.sbi").write_bytes(SBI)
self.assertEqual(self.prepare()[0], 1)
def test_invalid_records(self):
fixtures = [b"", b"SBI\0", SBI[:-1], b"BAD!" + SBI[4:],
SBI[:7] + b"\x02" + SBI[8:], SBI[:4] + b"\xfa" + SBI[5:],
SBI[:5] + b"\x60" + SBI[6:], SBI[:6] + b"\x75" + SBI[7:],
SBI[:4] + bytes(3) + SBI[7:], SBI + SBI[4:]]
for data in fixtures:
with self.subTest(data=data), self.assertRaises(dc.CompanionError):
dc.validate_sbi(data)
def test_unknown_iso_and_raw_conversion_preserve_supplied_bytes(self):
for kind in ["iso", "raw"]:
source = self.root / ("cooked.iso" if kind == "iso" else "raw.bin")
data = cooked_disc() if kind == "iso" else b"".join(
self.image.read_bytes()[i:i + 2352] + bytes(96)
for i in range(0, self.size, 2352))
source.write_bytes(data)
source.with_suffix(".sbi").write_bytes(SBI)
out = self.root / kind
code, message = self.prepare(source, out)
self.assertEqual(code, 0, message)
self.assertEqual((out / "renamed.sbi").read_bytes(), SBI)
report = json.loads((out / "renamed.disc-receipt.json").read_text())
self.assertEqual(report["subchannel"]["status"], "format_valid_unqualified")
def test_unknown_serial_is_not_protection_detection_and_cli_subprocess(self):
source = self.root / "unknown.bin"
source.write_bytes(self.image.read_bytes() + bytes(2352))
code, message = self.prepare(source)
self.assertEqual(code, 0, message)
self.assertIn("subchannel: unknown", message)
proc = subprocess.run([sys.executable, str(ROOT / "psxrecomp_cli.py"),
"verify-disc", "--config", str(self.config), "--disc", str(source),
"--json-progress"], capture_output=True, text=True)
self.assertEqual(proc.returncode, 0, proc.stderr)
events = [json.loads(line) for line in proc.stdout.splitlines()]
self.assertEqual(events[-1]["subchannel"]["status"], "unknown")
if __name__ == "__main__":
unittest.main()