637 lines
34 KiB
Swift
637 lines
34 KiB
Swift
import Foundation
|
|
@preconcurrency import Network
|
|
import Security
|
|
|
|
#if CAST_SESSION_CHECKS
|
|
struct CastTextTrack: Sendable {
|
|
let id: Int
|
|
let url: URL
|
|
let language: String
|
|
let label: String
|
|
}
|
|
struct CastMediaDescription: Sendable {
|
|
let url: URL
|
|
let contentType: String
|
|
let duration: Double
|
|
let live: Bool
|
|
let tracks: [CastTextTrack]
|
|
}
|
|
#endif
|
|
|
|
private func require(_ condition: @autoclosure () -> Bool, _ message: String) {
|
|
if !condition() { fatalError("FAIL: " + message) }
|
|
}
|
|
|
|
@MainActor
|
|
private func until(_ message: String, attempts: Int = 500, _ predicate: () -> Bool) async throws {
|
|
// An I/O watchdog, not a wall-clock date used to decide playback behavior.
|
|
for _ in 0..<attempts {
|
|
if predicate() { return }
|
|
try await Task.sleep(for: .milliseconds(10))
|
|
}
|
|
require(predicate(), message)
|
|
}
|
|
@MainActor
|
|
private final class CastCheckClock {
|
|
private struct Waiter {
|
|
let delay: Duration
|
|
let deadline: Duration
|
|
let continuation: CheckedContinuation<Void, any Error>
|
|
}
|
|
private var now: Duration = .zero
|
|
private var waiters: [UUID: Waiter] = [:]
|
|
var pendingCount: Int { waiters.count }
|
|
func hasDelay(_ delay: Duration) -> Bool { waiters.values.contains { $0.delay == delay } }
|
|
|
|
func sleep(_ delay: Duration) async throws {
|
|
try Task.checkCancellation()
|
|
let id = UUID()
|
|
try await withTaskCancellationHandler {
|
|
try await withCheckedThrowingContinuation { continuation in
|
|
waiters[id] = Waiter(delay: delay, deadline: now + delay, continuation: continuation)
|
|
}
|
|
} onCancel: {
|
|
Task { @MainActor [weak self] in
|
|
self?.waiters.removeValue(forKey: id)?.continuation.resume(throwing: CancellationError())
|
|
}
|
|
}
|
|
try Task.checkCancellation()
|
|
}
|
|
|
|
func advance(_ duration: Duration) {
|
|
now += duration
|
|
let due = waiters.filter { $0.value.deadline <= now }.map(\.key)
|
|
for id in due { waiters.removeValue(forKey: id)?.continuation.resume() }
|
|
}
|
|
}
|
|
|
|
|
|
@MainActor
|
|
private final class Receiver {
|
|
private let listener: NWListener
|
|
private var socket: NWConnection?
|
|
private var stream = CastWire.Stream()
|
|
private var sender = ""
|
|
private let transport = "receiver-app"
|
|
private var media: [String: Any] = [:]
|
|
private var session = 1
|
|
private var position = 0.0
|
|
private var playing = false
|
|
private var activeTracks: [Int] = []
|
|
var commands: [(String, [String: Any])] = []
|
|
var rejectLoad = false
|
|
var rejectLaunch = false
|
|
var dropLaunch = false
|
|
var dropLoad = false
|
|
var dropCommands = false
|
|
var rejectCommand = false
|
|
var holdSeek = false
|
|
var ignoreHeartbeat = false
|
|
var verifyRelay = false
|
|
var relayVerified = false
|
|
var failure: String?
|
|
private var heldSeek: [String: Any]?
|
|
private var started = false
|
|
private var volume = 1.0
|
|
var muted = false
|
|
var port: UInt16 { listener.port!.rawValue }
|
|
var device: CastDevice { .init(id: "loopback", name: "Loopback TV", host: "127.0.0.1", port: port) }
|
|
|
|
init(identity: SecIdentity) throws {
|
|
let tls = NWProtocolTLS.Options()
|
|
guard let identity = sec_identity_create(identity) else { throw CastError.protocolError }
|
|
sec_protocol_options_set_local_identity(tls.securityProtocolOptions, identity)
|
|
sec_protocol_options_set_min_tls_protocol_version(tls.securityProtocolOptions, .TLSv12)
|
|
let parameters = NWParameters(tls: tls, tcp: NWProtocolTCP.Options())
|
|
parameters.requiredLocalEndpoint = .hostPort(host: "127.0.0.1", port: .any)
|
|
listener = try NWListener(using: parameters)
|
|
}
|
|
|
|
func start() async throws {
|
|
listener.stateUpdateHandler = { [weak self] state in
|
|
Task { @MainActor in
|
|
switch state {
|
|
case .ready: self?.started = true
|
|
case .failed: self?.failure = "TLS listener failed"
|
|
default: break
|
|
}
|
|
}
|
|
}
|
|
listener.newConnectionHandler = { [weak self] socket in
|
|
Task { @MainActor in
|
|
guard let self else { socket.cancel(); return }
|
|
self.socket?.cancel()
|
|
self.socket = socket
|
|
self.stream = CastWire.Stream()
|
|
socket.stateUpdateHandler = { [weak self] state in
|
|
Task { @MainActor in
|
|
if case .ready = state { self?.receive(socket) }
|
|
}
|
|
}
|
|
socket.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
}
|
|
}
|
|
listener.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
try await until("TLS listener starts") { started || failure != nil }
|
|
require(failure == nil, "TLS listener succeeds")
|
|
}
|
|
|
|
func stop() { socket?.cancel(); listener.cancel() }
|
|
|
|
private func receive(_ socket: NWConnection) {
|
|
socket.receive(minimumIncompleteLength: 1, maximumLength: 16384) { [weak self] data, _, ended, error in
|
|
Task { @MainActor in
|
|
guard let self else { return }
|
|
do {
|
|
if let data {
|
|
for message in try self.stream.append(data) { try await self.handle(message) }
|
|
}
|
|
if !ended, error == nil { self.receive(socket) }
|
|
} catch { self.failure = "Simulated receiver protocol/relay failure: \(error.localizedDescription)" }
|
|
}
|
|
}
|
|
}
|
|
|
|
private func send(_ json: [String: Any], namespace: String, source: String? = nil, fragmented: Bool = false) throws {
|
|
let payload = try JSONSerialization.data(withJSONObject: json)
|
|
let frame = try CastWire.frame(.init(source: source ?? (namespace == CastWire.media ? transport : "receiver-0"),
|
|
destination: sender, namespace: namespace, payload: payload))
|
|
if fragmented {
|
|
// Split both the length prefix and protobuf, exercising genuine receive framing.
|
|
for chunk in [frame.prefix(2), frame.dropFirst(2).prefix(5), frame.dropFirst(7)] {
|
|
socket?.send(content: Data(chunk), completion: .contentProcessed { _ in })
|
|
}
|
|
} else { socket?.send(content: frame, completion: .contentProcessed { _ in }) }
|
|
}
|
|
|
|
private func receiverStatus(_ requestID: Int) throws {
|
|
try send(["type": "RECEIVER_STATUS", "requestId": requestID,
|
|
"status": ["applications": [["appId": "CC1AD845", "sessionId": "app-session", "transportId": transport]],
|
|
"volume": ["level": volume, "muted": muted]]], namespace: CastWire.receiver, fragmented: true)
|
|
}
|
|
|
|
func status(requestID: Int = 0, session override: Int? = nil, content: String? = nil,
|
|
state: String? = nil, idle: String? = nil) throws {
|
|
var media = self.media
|
|
if let content { media["contentId"] = content }
|
|
try send(["type": "MEDIA_STATUS", "requestId": requestID,
|
|
"status": [["mediaSessionId": override ?? session, "media": media,
|
|
"playerState": state ?? (playing ? "PLAYING" : "PAUSED"), "currentTime": position,
|
|
"idleReason": idle ?? "", "supportedMediaCommands": 4099, "activeTrackIds": activeTracks]]],
|
|
namespace: CastWire.media, fragmented: true)
|
|
}
|
|
|
|
func releaseSeek() throws {
|
|
guard let command = heldSeek else { return }
|
|
heldSeek = nil
|
|
position = command["currentTime"] as? Double ?? position
|
|
try status(requestID: command["requestId"] as? Int ?? 0)
|
|
}
|
|
|
|
func invalidFrame() {
|
|
socket?.send(content: Data([0, 1, 0, 1]), completion: .contentProcessed { _ in }) // 65537 > bound
|
|
}
|
|
|
|
func takeOverApplication() throws {
|
|
try send(["type": "RECEIVER_STATUS", "status": ["applications": []]], namespace: CastWire.receiver)
|
|
}
|
|
|
|
private func handle(_ message: CastWire.Message) async throws {
|
|
sender = message.source
|
|
let json = try JSONSerialization.jsonObject(with: message.payload) as! [String: Any]
|
|
let type = json["type"] as! String
|
|
let requestID = json["requestId"] as? Int ?? 0
|
|
commands.append((type, json))
|
|
if message.namespace == CastWire.heartbeat {
|
|
if type == "PING", !ignoreHeartbeat { try send(["type": "PONG"], namespace: CastWire.heartbeat) }
|
|
return
|
|
}
|
|
switch type {
|
|
case "CONNECT": break
|
|
case "CLOSE": break
|
|
case "LAUNCH":
|
|
require(json["appId"] as? String == "CC1AD845", "Default Media Receiver, no registration")
|
|
if dropLaunch { return }
|
|
if rejectLaunch { try send(["type": "LAUNCH_ERROR", "requestId": requestID], namespace: CastWire.receiver) }
|
|
else { try receiverStatus(requestID) }
|
|
case "LOAD":
|
|
if rejectLoad { try send(["type": "LOAD_FAILED", "requestId": requestID], namespace: CastWire.media); return }
|
|
if dropLoad { return }
|
|
media = json["media"] as! [String: Any]
|
|
session += 1
|
|
playing = json["autoplay"] as? Bool ?? true
|
|
position = json["currentTime"] as? Double ?? 0
|
|
activeTracks = json["activeTrackIds"] as? [Int] ?? []
|
|
if verifyRelay { try await fetchRelay() }
|
|
try status(requestID: requestID)
|
|
case "PLAY", "PAUSE", "SEEK", "GET_STATUS", "EDIT_TRACKS_INFO":
|
|
if dropCommands { return }
|
|
if rejectCommand {
|
|
try send(["type": "INVALID_PLAYER_STATE", "requestId": requestID], namespace: CastWire.media)
|
|
return
|
|
}
|
|
require(json["mediaSessionId"] as? Int == session, "Controls address the acquired media session")
|
|
if type == "PLAY" { playing = true }
|
|
if type == "PAUSE" { playing = false }
|
|
if type == "EDIT_TRACKS_INFO" { activeTracks = json["activeTrackIds"] as? [Int] ?? [] }
|
|
if type == "SEEK" {
|
|
if holdSeek { heldSeek = json; try status(); return }
|
|
position = json["currentTime"] as? Double ?? position
|
|
}
|
|
try status(requestID: requestID)
|
|
case "SET_VOLUME":
|
|
require(message.namespace == CastWire.receiver, "Volume controls device, not stream")
|
|
volume = (json["volume"] as? [String: Any])?["level"] as? Double ?? volume
|
|
muted = (json["volume"] as? [String: Any])?["muted"] as? Bool ?? muted
|
|
try receiverStatus(requestID)
|
|
case "STOP":
|
|
require(message.namespace == CastWire.media, "Disconnect does not stop another sender's application")
|
|
require(json["mediaSessionId"] as? Int == session, "Disconnect stops only owned media")
|
|
default: failure = "Unexpected command \(type)"
|
|
}
|
|
}
|
|
|
|
private func fetchRelay() async throws {
|
|
let url = URL(string: media["contentId"] as! String)!
|
|
let config = URLSessionConfiguration.ephemeral
|
|
config.timeoutIntervalForRequest = 30
|
|
let session = URLSession(configuration: config)
|
|
defer { session.invalidateAndCancel() }
|
|
func fetch(_ url: URL) async throws -> Data {
|
|
let (data, response) = try await session.data(from: url)
|
|
require((response as? HTTPURLResponse)?.statusCode == 200, "Receiver fetch succeeds")
|
|
require(!data.isEmpty, "Relay returns real bytes")
|
|
return data
|
|
}
|
|
var playlistURL = url
|
|
var playlist = try await fetch(playlistURL)
|
|
var body = String(data: playlist, encoding: .utf8) ?? ""
|
|
require(body.hasPrefix("#EXTM3U"), "LOAD is playable HLS")
|
|
for _ in 0..<3 {
|
|
guard let uri = body.split(separator: "\n").first(where: { !$0.hasPrefix("#") && !$0.isEmpty }),
|
|
let next = URL(string: String(uri).trimmingCharacters(in: .whitespacesAndNewlines), relativeTo: playlistURL)?.absoluteURL else {
|
|
fatalError("FAIL: playlist has a media resource")
|
|
}
|
|
let data = try await fetch(next)
|
|
if let text = String(data: data, encoding: .utf8), text.hasPrefix("#EXTM3U") {
|
|
playlistURL = next; playlist = data; body = text; continue
|
|
}
|
|
require(data.count >= 188 && data.first == 0x47 || data.count >= 12 &&
|
|
["ftyp", "styp", "moof"].contains(String(data: data[4..<8], encoding: .ascii) ?? ""),
|
|
"Relay serves MPEG-TS or fragmented MP4, not placeholder content")
|
|
for track in media["tracks"] as? [[String: Any]] ?? [] {
|
|
if let textURL = (track["trackContentId"] as? String).flatMap(URL.init(string:)) {
|
|
let text = try await fetch(textURL)
|
|
require(String(data: text, encoding: .utf8)?.hasPrefix("WEBVTT") == true, "Receiver subtitles are WebVTT")
|
|
}
|
|
}
|
|
relayVerified = true
|
|
return
|
|
}
|
|
fatalError("FAIL: receiver playlist resolves to real media")
|
|
}
|
|
}
|
|
|
|
@MainActor
|
|
private func testIdentity() throws -> (SecIdentity, URL) {
|
|
let directory = FileManager.default.temporaryDirectory.appendingPathComponent("terebi-cast-check-" + UUID().uuidString)
|
|
try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
|
|
func openssl(_ arguments: [String]) throws {
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: "/usr/bin/openssl")
|
|
process.arguments = arguments
|
|
process.standardOutput = FileHandle.nullDevice
|
|
process.standardError = FileHandle.nullDevice
|
|
try process.run()
|
|
process.waitUntilExit()
|
|
require(process.terminationStatus == 0, "Creates ephemeral self-signed TLS test identity")
|
|
}
|
|
let key = directory.appendingPathComponent("key.pem").path
|
|
let certificate = directory.appendingPathComponent("cert.pem").path
|
|
let bundle = directory.appendingPathComponent("identity.p12")
|
|
try openssl(["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key,
|
|
"-out", certificate, "-days", "1", "-subj", "/CN=Cast loopback"])
|
|
try openssl(["pkcs12", "-export", "-inkey", key, "-in", certificate, "-out", bundle.path,
|
|
"-passout", "pass:cast-check", "-keypbe", "PBE-SHA1-3DES", "-certpbe", "PBE-SHA1-3DES", "-macalg", "sha1"])
|
|
let data = try Data(contentsOf: bundle)
|
|
var result: CFArray?
|
|
let status = SecPKCS12Import(data as CFData,
|
|
[kSecImportExportPassphrase as String: "cast-check"] as CFDictionary, &result)
|
|
require(status == errSecSuccess, "Imports ephemeral TLS identity")
|
|
let item = (result! as! [[String: Any]])[0]
|
|
return (item[kSecImportItemIdentity as String] as! SecIdentity, directory)
|
|
}
|
|
|
|
@main
|
|
@MainActor
|
|
struct CastSessionCheck {
|
|
static func codec() throws {
|
|
let payload = Data("{\"type\":\"PING\"}".utf8)
|
|
let message = CastWire.Message(source: "sender-0", destination: "receiver-0",
|
|
namespace: CastWire.heartbeat, payload: payload)
|
|
let frame = try CastWire.frame(message)
|
|
var stream = CastWire.Stream()
|
|
var decoded: [CastWire.Message] = []
|
|
for byte in frame { decoded += try stream.append(Data([byte])) }
|
|
require(decoded.count == 1 && decoded[0].payload == payload, "Byte-fragmented frame round trips")
|
|
let pair = try stream.append(frame + frame)
|
|
require(pair.count == 2, "Coalesced frames decode")
|
|
for malformed in [Data([0, 0, 0, 0]), Data([0, 1, 0, 1]), Data([255, 255, 255, 255])] {
|
|
do { var parser = CastWire.Stream(); _ = try parser.append(malformed); fatalError("FAIL: invalid length accepted") }
|
|
catch CastError.protocolError { }
|
|
}
|
|
for malformed in [Data([8, 128]), Data([8, 0, 18, 127]), Data([15]), Data(repeating: 255, count: 12)] {
|
|
do { _ = try CastWire.decode(malformed); fatalError("FAIL: malformed protobuf accepted") }
|
|
catch CastError.protocolError { }
|
|
}
|
|
do {
|
|
_ = try CastWire.frame(.init(source: "s", destination: "r", namespace: "m",
|
|
payload: Data(repeating: 65, count: CastWire.maximumFrame)))
|
|
fatalError("FAIL: oversized outbound frame accepted")
|
|
} catch CastError.protocolError { }
|
|
}
|
|
|
|
static func media(_ suffix: String = "first") -> CastMediaDescription {
|
|
.init(url: URL(string: "http://127.0.0.1:9123/session-token/\(suffix).m3u8")!,
|
|
contentType: "application/x-mpegURL", duration: 120, live: false,
|
|
tracks: [.init(id: 17, url: URL(string: "http://127.0.0.1:9123/session-token/subtitle.vtt")!,
|
|
language: "en", label: "English")])
|
|
}
|
|
|
|
static func behavior(identity: SecIdentity) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.muted = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
var statuses: [CastStatus] = []
|
|
var errors: [String] = []
|
|
var disconnected = false
|
|
session.onStatus = { statuses.append($0) }
|
|
session.onError = { errors.append($0) }
|
|
session.onDisconnect = { _ in disconnected = true }
|
|
try await session.connect(to: peer.device)
|
|
require(peer.commands.first?.0 == "CONNECT", "TLS platform channel connects before launching")
|
|
require(!session.setPlaying(true), "No media control before load acknowledgement")
|
|
try await session.load(media(), title: "Check", position: 9.25, playing: false, subtitle: 0)
|
|
require(statuses.last?.position == 9.25 && statuses.last?.playing == false && statuses.last?.subtitle == 0,
|
|
"LOAD is receiver-authoritative and subtitle IDs map to indices")
|
|
require(statuses.last?.volume == 0, "Pre-existing device mute is reflected in controls")
|
|
require(session.setPlaying(true) && session.setPlaying(false), "Rapid opposite commands accepted")
|
|
try await until("PLAY and PAUSE acknowledged in order") {
|
|
peer.commands.filter { ["PLAY", "PAUSE"].contains($0.0) }.count == 2 && statuses.last?.playing == false
|
|
}
|
|
require(peer.commands.filter { ["PLAY", "PAUSE"].contains($0.0) }.map(\.0) == ["PLAY", "PAUSE"], "Ordered controls")
|
|
let beforeBurst = statuses.count
|
|
require(session.setPlaying(true) && session.setPlaying(false) && session.setPlaying(true),
|
|
"Three rapid intents are accepted")
|
|
try await until("Latest playback intent is acknowledged") {
|
|
statuses.last?.playing == true && statuses.last?.playbackPending == false
|
|
}
|
|
require(statuses.dropFirst(beforeBurst).contains { $0.playing && $0.playbackPending },
|
|
"An older matching PLAY cannot settle the latest playback intent")
|
|
require(session.setPlaying(false), "Pause before seek checks")
|
|
try await until("Pause settles") { statuses.last?.playing == false && statuses.last?.playbackPending == false }
|
|
peer.holdSeek = true
|
|
let beforeSeek = statuses.count
|
|
require(session.seek(24.25), "SEEK accepted")
|
|
try await until("Old status arrives while SEEK pending") {
|
|
peer.commands.contains { $0.0 == "SEEK" } && statuses.count >= beforeSeek + 2
|
|
}
|
|
require(statuses.last?.seekPending == true && statuses.last?.position == 9.25,
|
|
"Old status cannot clear an unacknowledged seek or invent position")
|
|
try await until("Heartbeat is waiting on the controlled clock") { clock.hasDelay(.seconds(5)) }
|
|
clock.advance(.seconds(5))
|
|
try await until("Newer status poll replies before held SEEK") {
|
|
peer.commands.contains { $0.0 == "GET_STATUS" } && statuses.last?.seekPending == true
|
|
}
|
|
try peer.releaseSeek()
|
|
try await until("SEEK request correlation") { statuses.last?.seekPending == false && statuses.last?.position == 24.25 }
|
|
require(session.setVolume(0.3), "Device volume accepted")
|
|
try await until("Device volume acknowledged") { statuses.last?.volume == 0.3 }
|
|
require(session.setSubtitle(nil), "Subtitle off accepted")
|
|
try await until("Subtitle off acknowledged") { statuses.last?.subtitle == nil }
|
|
let oldSession = 2
|
|
try await session.load(media("replacement"), title: "Replacement", position: 3, playing: true, subtitle: nil)
|
|
let statusCount = statuses.count
|
|
try peer.status(session: oldSession, content: media().url.absoluteString)
|
|
try peer.status()
|
|
try await until("Valid status after stale frame provides an ordered receiver barrier") {
|
|
statuses.count >= statusCount + 1
|
|
}
|
|
require(statuses.count == statusCount + 1 && !disconnected, "Only the valid barrier status is published")
|
|
try peer.status(state: "IDLE", idle: "FINISHED")
|
|
try await until("Receiver reports natural end") { statuses.last?.ended == true }
|
|
let beforeReplay = peer.commands.filter { $0.0 == "LOAD" }.count
|
|
require(session.setPlaying(true) && session.setPlaying(false), "Ended replay and rapid pause accepted")
|
|
try await until("Ended replay uses LOAD from zero then honors pause") {
|
|
peer.commands.filter { $0.0 == "LOAD" }.count == beforeReplay + 1 &&
|
|
statuses.last?.position == 0 && statuses.last?.playing == false && statuses.last?.ended == false
|
|
}
|
|
require(peer.commands.last(where: { $0.0 == "LOAD" })?.1["currentTime"] as? Double == 0,
|
|
"Default receiver ended media reloaded, not invalid PLAY")
|
|
require(!session.seek(.nan) && !session.setVolume(2) && !session.setSubtitle(99), "Invalid command input rejected")
|
|
session.disconnect()
|
|
try await until("Owned media STOP arrives") { peer.commands.contains { $0.0 == "STOP" } }
|
|
require(disconnected && errors.isEmpty && peer.failure == nil, "Safe disconnect without spurious cancellation errors")
|
|
}
|
|
|
|
static func connectionDeadline() async throws {
|
|
let listener = try NWListener(using: .tcp, on: .any)
|
|
@MainActor final class ListenerState {
|
|
var accepted: [NWConnection] = []
|
|
var listening = false
|
|
}
|
|
let state = ListenerState()
|
|
listener.stateUpdateHandler = { event in
|
|
Task { @MainActor in if case .ready = event { state.listening = true } }
|
|
}
|
|
listener.newConnectionHandler = { socket in
|
|
Task { @MainActor in
|
|
state.accepted.append(socket)
|
|
// Accept TCP but never answer the client's TLS handshake.
|
|
socket.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
}
|
|
}
|
|
listener.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
defer { for socket in state.accepted { socket.cancel() }; listener.cancel() }
|
|
try await until("Incomplete TLS listener starts") { state.listening }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
let task = Task {
|
|
try await session.connect(to: .init(id: "stall", name: "Stalled TLS", host: "127.0.0.1",
|
|
port: listener.port!.rawValue))
|
|
}
|
|
try await until("Connection handshake waits on frozen clock") {
|
|
!state.accepted.isEmpty && clock.hasDelay(.seconds(12))
|
|
}
|
|
clock.advance(.seconds(12))
|
|
do { try await task.value; fatalError("FAIL: stalled TLS handshake succeeded") }
|
|
catch CastError.timeout { }
|
|
try await until("Connection deadline cancels all timer handles") { clock.pendingCount == 0 }
|
|
}
|
|
|
|
static func failures(identity: SecIdentity) async throws {
|
|
for mode in ["launch", "launch-timeout", "load", "deadline", "command", "rejection", "cancel", "frame",
|
|
"ownership", "media-error", "media-cancel", "media-interrupt", "heartbeat", "queue"] {
|
|
let peer = try Receiver(identity: identity)
|
|
try await peer.start()
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(requestTimeout: .milliseconds(400), loadTimeout: .milliseconds(400),
|
|
heartbeatInterval: mode == "heartbeat" ? .milliseconds(100) : .seconds(5),
|
|
sleep: { try await clock.sleep($0) })
|
|
var disconnected = false
|
|
var errors: [String] = []
|
|
session.onDisconnect = { _ in disconnected = true }
|
|
session.onError = { errors.append($0) }
|
|
if mode == "launch" { peer.rejectLaunch = true }
|
|
if mode == "launch-timeout" { peer.dropLaunch = true }
|
|
if mode == "heartbeat" { peer.ignoreHeartbeat = true }
|
|
if mode == "launch" {
|
|
do { try await session.connect(to: peer.device); fatalError("FAIL: rejected launch succeeded") }
|
|
catch CastError.rejected { }
|
|
} else if mode == "launch-timeout" {
|
|
let task = Task { try await session.connect(to: peer.device) }
|
|
try await until("Receiver launch waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "LAUNCH" } && clock.hasDelay(.milliseconds(400))
|
|
}
|
|
clock.advance(.milliseconds(400))
|
|
do { try await task.value; fatalError("FAIL: unanswered launch succeeded") }
|
|
catch CastError.timeout { require(disconnected, "Launch deadline releases channel") }
|
|
} else {
|
|
try await session.connect(to: peer.device)
|
|
try await until("Handshake deadline handles cancelled") { clock.pendingCount == 1 }
|
|
peer.rejectLoad = mode == "load"
|
|
peer.dropLoad = mode == "deadline" || mode == "cancel"
|
|
if mode == "load" || mode == "deadline" {
|
|
let task = Task { try await session.load(media(), title: "Failure", position: 0, playing: true, subtitle: nil) }
|
|
if mode == "deadline" {
|
|
try await until("LOAD waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "LOAD" } && clock.hasDelay(.milliseconds(400))
|
|
}
|
|
clock.advance(.milliseconds(400))
|
|
}
|
|
do { try await task.value; fatalError("FAIL: failed load succeeded") }
|
|
catch CastError.rejected { require(mode == "load", "LOAD rejection correlated") }
|
|
catch CastError.timeout { require(mode == "deadline", "LOAD deadline bounded") }
|
|
} else if mode == "cancel" {
|
|
let task = Task { try await session.load(media(), title: "Cancel", position: 0, playing: true, subtitle: nil) }
|
|
try await until("LOAD sent before cancellation") { peer.commands.contains { $0.0 == "LOAD" } }
|
|
task.cancel()
|
|
do { try await task.value; fatalError("FAIL: cancelled LOAD succeeded") }
|
|
catch { require(disconnected, "LOAD cancellation releases channel") }
|
|
} else {
|
|
try await session.load(media(), title: "Check", position: 0, playing: false, subtitle: nil)
|
|
switch mode {
|
|
case "command":
|
|
peer.dropCommands = true
|
|
require(session.seek(12), "Unanswered SEEK sent")
|
|
try await until("SEEK waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "SEEK" } && clock.hasDelay(.milliseconds(400))
|
|
}
|
|
clock.advance(.milliseconds(400))
|
|
try await until("Command deadline reports recoverable error") { !errors.isEmpty }
|
|
require(!disconnected, "Command timeout is not channel failure")
|
|
case "rejection":
|
|
peer.rejectCommand = true
|
|
require(session.setPlaying(true), "Rejected command sent")
|
|
try await until("Command rejection reports recoverable error") { !errors.isEmpty }
|
|
require(!disconnected, "Command rejection is not channel failure")
|
|
case "media-error": try peer.status(state: "IDLE", idle: "ERROR")
|
|
case "media-cancel": try peer.status(state: "IDLE", idle: "CANCELLED")
|
|
case "media-interrupt": try peer.status(state: "IDLE", idle: "INTERRUPTED")
|
|
case "frame": peer.invalidFrame()
|
|
case "ownership": try peer.takeOverApplication()
|
|
case "heartbeat":
|
|
try await until("Heartbeat tick is scheduled on frozen clock") { clock.hasDelay(.milliseconds(100)) }
|
|
clock.advance(.milliseconds(100))
|
|
try await until("PING sent and next heartbeat tick scheduled") {
|
|
peer.commands.contains { $0.0 == "PING" } && clock.hasDelay(.milliseconds(100))
|
|
}
|
|
clock.advance(.milliseconds(100))
|
|
case "queue":
|
|
peer.dropCommands = true
|
|
var accepted = 0
|
|
for _ in 0..<40 { if session.setPlaying(true) { accepted += 1 } }
|
|
require(accepted == 32 && !errors.isEmpty, "Pending command resource bound enforced")
|
|
default: break
|
|
}
|
|
if ["frame", "ownership", "media-error", "media-cancel", "media-interrupt", "heartbeat"].contains(mode) {
|
|
try await until("\(mode) causes bounded channel failure") { disconnected }
|
|
}
|
|
}
|
|
}
|
|
session.disconnect(stopReceiver: false)
|
|
try await until("Cancelled timer handles released") { clock.pendingCount == 0 }
|
|
peer.stop()
|
|
}
|
|
}
|
|
|
|
static func playback(identity: SecIdentity, executable: String, file: String) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.verifyRelay = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: executable)
|
|
process.arguments = [file, String(peer.port)]
|
|
try process.run()
|
|
defer { if process.isRunning { process.terminate() } }
|
|
try await until("Integrated playback subprocess completes", attempts: 18000) { !process.isRunning || peer.failure != nil }
|
|
require(peer.failure == nil, peer.failure ?? "Simulated receiver accepts integration flow")
|
|
require(process.terminationStatus == 0, "Player model integration check succeeds")
|
|
try await until("Disconnect STOP reaches receiver") { peer.commands.contains { $0.0 == "STOP" } }
|
|
let load = peer.commands.first { $0.0 == "LOAD" }!.1
|
|
require(load["currentTime"] as? Double == 9.25 && load["autoplay"] as? Bool == false, "Acknowledged paused handoff")
|
|
require(peer.relayVerified, "Receiver fetches real relay HLS and media bytes")
|
|
let controls = peer.commands.filter { ["PLAY", "SEEK", "PAUSE", "STOP"].contains($0.0) }
|
|
require(controls.map(\.0) == ["PLAY", "SEEK", "SEEK", "PAUSE", "STOP"], "Integrated controls ordered")
|
|
require(controls[1].1["currentTime"] as? Double == 14.25 &&
|
|
controls[2].1["currentTime"] as? Double == 24.25, "Rapid skips accumulate from the pending target")
|
|
}
|
|
|
|
static func realPeer(_ arguments: [String]) async throws {
|
|
guard let port = UInt16(arguments[3]), let url = URL(string: arguments[4]) else {
|
|
fatalError("Usage: cast-session-check --peer HOST PORT MEDIA_URL CONTENT_TYPE")
|
|
}
|
|
let session = CastSession()
|
|
var status: CastStatus?
|
|
var error: String?
|
|
session.onStatus = { status = $0 }
|
|
session.onDisconnect = { error = $0 }
|
|
session.onError = { error = $0 }
|
|
defer { session.disconnect() }
|
|
try await session.connect(to: .init(id: "smoke", name: "Smoke receiver", host: arguments[2], port: port))
|
|
try await session.load(.init(url: url, contentType: arguments[5], duration: 0, live: false, tracks: []),
|
|
title: "Terebi Cast smoke", position: 0, playing: false, subtitle: nil)
|
|
require(status != nil, "Real receiver acknowledges LOAD")
|
|
require(session.setPlaying(true), "Real receiver PLAY accepted")
|
|
try await until("Real receiver plays", attempts: 1500) { status?.playing == true || error != nil }
|
|
require(error == nil && status?.playing == true, "Real receiver acknowledges PLAY")
|
|
require(session.setPlaying(false), "Real receiver PAUSE accepted")
|
|
try await until("Real receiver pauses", attempts: 1500) { status?.playing == false || error != nil }
|
|
require(error == nil && status?.playing == false, "Real receiver acknowledges PAUSE")
|
|
print("PASS: real peer TLS Default Media Receiver LOAD/PLAY/PAUSE (URLs intentionally omitted)")
|
|
}
|
|
|
|
static func main() async throws {
|
|
if CommandLine.arguments.count == 6, CommandLine.arguments[1] == "--peer" {
|
|
try await realPeer(CommandLine.arguments)
|
|
return
|
|
}
|
|
try codec()
|
|
let (identity, directory) = try testIdentity()
|
|
defer { try? FileManager.default.removeItem(at: directory) }
|
|
if CommandLine.arguments.count == 4, CommandLine.arguments[1] == "--playback" {
|
|
try await playback(identity: identity, executable: CommandLine.arguments[2], file: CommandLine.arguments[3])
|
|
print("PASS: native TLS receiver fetches relay and acknowledges PlayerModel handoff/control/disconnect")
|
|
} else {
|
|
try await behavior(identity: identity)
|
|
try await failures(identity: identity)
|
|
try await connectionDeadline()
|
|
print("PASS: bounded Cast framing, self-signed TLS launch/load/control, correlation, stale ownership, queue, deadlines, heartbeat, cancellation")
|
|
}
|
|
}
|
|
}
|