use std::collections::{BTreeMap, BTreeSet};
use std::time::Duration;
use dvb_ci::objects::ca_pmt::CaPmtCmdId;
use dvb_ci::objects::ca_pmt_reply::CaEnable;
use dvb_si::descriptors::DescriptorLoop;
use dvb_si::descriptors::ca::TAG as CA_DESCRIPTOR_TAG;
use dvb_si::tables::cat::CatCaEntry;
use dvb_si::tables::pmt::PmtSection;
pub const REQUERY_DEFAULT: Duration = Duration::from_secs(10);
const PCR_PID_NONE: u16 = 0x1FFF;
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum CaError {
#[error("PMT for program_number {program_number} has no CA_descriptor at program or ES level")]
NoCaDescriptor {
program_number: u16,
},
#[error("ca_pmt send failed: {0}")]
Io(#[from] std::io::Error),
#[error("CAT CA_descriptor parse failed: {0}")]
Cat(#[from] dvb_si::error::Error),
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct ManagedService {
pub es_pids: Vec<u16>,
pub ca_pids: Vec<u16>,
pub pcr_pid: u16,
pub cmd: CaPmtCmdId,
pub last_ca_enable: Option<CaEnable>,
pub(crate) last_descrambling_ok: bool,
pub(crate) built_ca_pmt: Vec<u8>,
pub(crate) pmt_raw: Vec<u8>,
}
#[derive(Debug, Clone)]
pub struct ManagedCa {
services: BTreeMap<u16, ManagedService>,
requery_interval: Duration,
since: Duration,
cat_emm_pids: BTreeMap<u16, u16>,
cam_caids: BTreeSet<u16>,
emm_pids: Vec<u16>,
descramble_pids: Vec<u16>,
ca_pids: Vec<u16>,
}
impl Default for ManagedCa {
fn default() -> Self {
Self {
services: BTreeMap::new(),
requery_interval: REQUERY_DEFAULT,
since: Duration::ZERO,
cat_emm_pids: BTreeMap::new(),
cam_caids: BTreeSet::new(),
emm_pids: Vec::new(),
descramble_pids: Vec::new(),
ca_pids: Vec::new(),
}
}
}
impl ManagedCa {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn services(&self) -> &BTreeMap<u16, ManagedService> {
&self.services
}
#[must_use]
pub fn requery_interval(&self) -> Duration {
self.requery_interval
}
pub(crate) fn set_requery_interval(&mut self, interval: Duration) {
self.requery_interval = interval;
self.since = Duration::ZERO;
}
#[must_use]
pub(crate) fn is_empty(&self) -> bool {
self.services.is_empty()
}
pub(crate) fn record(&mut self, program_number: u16, service: ManagedService) {
self.services.insert(program_number, service);
self.recompute_service_pids();
}
pub(crate) fn remove(&mut self, program_number: u16) -> bool {
let removed = self.services.remove(&program_number).is_some();
if removed {
self.recompute_service_pids();
}
removed
}
pub(crate) fn clear(&mut self) {
self.services.clear();
self.cat_emm_pids.clear();
self.cam_caids.clear();
self.emm_pids.clear();
self.descramble_pids.clear();
self.ca_pids.clear();
self.since = Duration::ZERO;
}
#[must_use]
pub fn emm_pids(&self) -> &[u16] {
&self.emm_pids
}
#[must_use]
pub fn descramble_pids(&self) -> &[u16] {
&self.descramble_pids
}
#[must_use]
pub fn ca_pids(&self) -> &[u16] {
&self.ca_pids
}
#[must_use]
pub fn required_pids(&self) -> Vec<u16> {
let mut pids: BTreeSet<u16> = BTreeSet::new();
pids.extend(self.descramble_pids.iter().copied());
pids.extend(self.ca_pids.iter().copied());
pids.extend(self.emm_pids.iter().copied());
for service in self.services.values() {
if service.pcr_pid != PCR_PID_NONE {
pids.insert(service.pcr_pid);
}
}
pids.into_iter().collect()
}
pub(crate) fn set_cat(&mut self, entries: &[CatCaEntry]) {
self.cat_emm_pids = entries.iter().map(|e| (e.ca_system_id, e.ca_pid)).collect();
self.recompute_emm_pids();
}
pub(crate) fn set_cam_caids(&mut self, caids: BTreeSet<u16>) {
self.cam_caids = caids;
self.recompute_emm_pids();
}
fn recompute_emm_pids(&mut self) {
let pids: BTreeSet<u16> = self
.cat_emm_pids
.iter()
.filter(|(caid, _)| self.cam_caids.contains(caid))
.map(|(_, pid)| *pid)
.collect();
self.emm_pids = pids.into_iter().collect();
}
pub(crate) fn tick(&mut self, elapsed: Duration) -> bool {
if self.requery_interval.is_zero() || self.services.is_empty() {
return false;
}
self.since += elapsed;
if self.since >= self.requery_interval {
self.since = Duration::ZERO;
true
} else {
false
}
}
pub(crate) fn record_reply(
&mut self,
program_number: u16,
ca_enable: Option<CaEnable>,
descrambling_ok: bool,
) -> Option<(CaEnable, bool)> {
let service = self.services.get_mut(&program_number)?;
let prev = (service.last_ca_enable, service.last_descrambling_ok);
service.last_ca_enable = ca_enable;
service.last_descrambling_ok = descrambling_ok;
match ca_enable {
Some(v) if prev != (ca_enable, descrambling_ok) => Some((v, descrambling_ok)),
_ => None,
}
}
fn recompute_service_pids(&mut self) {
let mut descramble: BTreeSet<u16> = BTreeSet::new();
let mut ca: BTreeSet<u16> = BTreeSet::new();
for service in self.services.values() {
descramble.extend(service.es_pids.iter().copied());
ca.extend(service.ca_pids.iter().copied());
}
self.descramble_pids = descramble.into_iter().collect();
self.ca_pids = ca.into_iter().collect();
}
}
fn has_ca_descriptor(loop_: &DescriptorLoop<'_>) -> bool {
loop_.raw_tags().any(|(tag, _)| tag == CA_DESCRIPTOR_TAG)
}
const CA_PID_BODY_OFFSET: usize = 2;
const CA_PID_FIELD_LEN: usize = 2;
const CA_PID_HIGH_MASK: u8 = 0x1F;
fn ca_pid_of(body: &[u8]) -> Option<u16> {
let field = body.get(CA_PID_BODY_OFFSET..CA_PID_BODY_OFFSET + CA_PID_FIELD_LEN)?;
Some((u16::from(field[0] & CA_PID_HIGH_MASK) << 8) | u16::from(field[1]))
}
fn ca_pids_in(loop_: &DescriptorLoop<'_>) -> Vec<u16> {
loop_
.raw_tags()
.filter(|(tag, _)| *tag == CA_DESCRIPTOR_TAG)
.filter_map(|(_, body)| ca_pid_of(body))
.collect()
}
pub(crate) fn pmt_has_ca(pmt: &PmtSection<'_>) -> bool {
has_ca_descriptor(&pmt.program_info)
|| pmt.streams.iter().any(|s| has_ca_descriptor(&s.es_info))
}
pub(crate) fn service_of(
pmt: &PmtSection<'_>,
cmd: CaPmtCmdId,
built_ca_pmt: Vec<u8>,
pmt_raw: Vec<u8>,
) -> ManagedService {
let mut ca_pids = ca_pids_in(&pmt.program_info);
for s in &pmt.streams {
ca_pids.extend(ca_pids_in(&s.es_info));
}
ManagedService {
es_pids: pmt.streams.iter().map(|s| s.elementary_pid).collect(),
ca_pids,
pcr_pid: pmt.pcr_pid,
cmd,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt,
pmt_raw,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn new_managed_ca_is_empty_at_default_cadence() {
let m = ManagedCa::new();
assert!(m.is_empty());
assert!(m.services().is_empty());
assert_eq!(m.requery_interval(), REQUERY_DEFAULT);
}
#[test]
fn record_tracks_the_service() {
let mut m = ManagedCa::new();
let svc = ManagedService {
es_pids: vec![0x100, 0x101],
ca_pids: vec![0x0064],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![0xAA, 0xBB],
pmt_raw: vec![0x02, 0x00],
};
m.record(7, svc.clone());
assert!(!m.is_empty());
assert_eq!(m.services().get(&7), Some(&svc));
}
#[test]
fn ca_error_no_ca_descriptor_displays_program_number() {
let e = CaError::NoCaDescriptor { program_number: 42 };
assert!(e.to_string().contains("42"));
}
#[test]
fn set_requery_interval_updates_and_resets_accumulator() {
let mut m = ManagedCa::new();
assert_eq!(m.requery_interval(), REQUERY_DEFAULT);
m.set_requery_interval(Duration::from_secs(3));
assert_eq!(m.requery_interval(), Duration::from_secs(3));
}
#[test]
fn tick_fires_once_interval_elapses_and_resets() {
let mut m = ManagedCa::new();
m.set_requery_interval(Duration::from_secs(5));
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert!(
!m.tick(Duration::from_secs(3)),
"before the interval: no fire"
);
assert!(
m.tick(Duration::from_secs(3)),
"crossing the interval: fires"
);
assert!(!m.tick(Duration::from_secs(1)), "since resets after firing");
}
#[test]
fn tick_disabled_at_zero_interval_never_fires() {
let mut m = ManagedCa::new();
m.set_requery_interval(Duration::ZERO);
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert!(!m.tick(Duration::from_secs(1000)));
}
#[test]
fn tick_with_no_active_services_never_fires() {
let mut m = ManagedCa::new();
m.set_requery_interval(Duration::from_secs(1));
assert!(!m.tick(Duration::from_secs(1000)));
}
#[test]
fn record_reply_first_ever_some_establishes_baseline_and_reports() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
let out = m.record_reply(1, Some(CaEnable::NotPossibleNoEntitlement), false);
assert_eq!(out, Some((CaEnable::NotPossibleNoEntitlement, false)));
}
#[test]
fn record_reply_unchanged_status_does_not_re_fire() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert!(m.record_reply(1, Some(CaEnable::Possible), true).is_some());
assert_eq!(m.record_reply(1, Some(CaEnable::Possible), true), None);
}
#[test]
fn record_reply_none_never_fires_but_updates_last() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert!(m.record_reply(1, Some(CaEnable::Possible), true).is_some());
assert_eq!(m.record_reply(1, None, false), None);
assert_eq!(
m.record_reply(1, Some(CaEnable::Possible), true),
Some((CaEnable::Possible, true))
);
}
#[test]
fn record_reply_unknown_program_is_a_no_op() {
let mut m = ManagedCa::new();
assert_eq!(m.record_reply(99, Some(CaEnable::Possible), true), None);
}
#[test]
fn remove_drops_tracked_service_and_recomputes_descramble_pids_false_for_untracked() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x100, 0x101],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
m.record(
2,
ManagedService {
es_pids: vec![0x200],
ca_pids: vec![0x65],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert!(
!m.remove(99),
"removing an untracked program_number must return false"
);
assert_eq!(
m.services().len(),
2,
"an untracked remove must not disturb the tracked set"
);
assert!(
m.remove(1),
"removing a tracked program_number must return true"
);
assert!(m.services().get(&1).is_none());
assert_eq!(
m.descramble_pids(),
&[0x200],
"descramble_pids must recompute (drop program 1's PIDs) after remove"
);
}
#[test]
fn clear_resets_module_state_but_preserves_requery_interval() {
use dvb_si::tables::cat::CatCaEntry;
let mut m = ManagedCa::new();
m.set_requery_interval(Duration::from_secs(3));
m.record(
1,
ManagedService {
es_pids: vec![0x100],
ca_pids: vec![0x64],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
m.set_cat(&[CatCaEntry {
ca_system_id: 0x0648,
ca_pid: 0x1FF0,
private_data: Vec::new(),
}]);
m.set_cam_caids([0x0648].into_iter().collect());
assert!(!m.emm_pids().is_empty(), "precondition: emm_pids populated");
assert!(
!m.descramble_pids().is_empty(),
"precondition: descramble_pids populated"
);
m.clear();
assert!(m.services().is_empty(), "services must be cleared");
assert!(m.emm_pids().is_empty(), "emm_pids must be cleared");
assert!(
m.descramble_pids().is_empty(),
"descramble_pids must be cleared"
);
assert!(m.ca_pids().is_empty(), "ca_pids must be cleared");
assert_eq!(
m.requery_interval(),
Duration::from_secs(3),
"requery_interval is host config, must survive clear()"
);
}
#[test]
fn ca_pids_is_the_dedup_sorted_union_of_active_services_and_required_pids_unions_all_three() {
use dvb_si::tables::cat::CatCaEntry;
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x0100, 0x0101],
ca_pids: vec![0x0064, 0x0065],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
m.record(
2,
ManagedService {
es_pids: vec![0x0200],
ca_pids: vec![0x0065, 0x0066],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert_eq!(
m.ca_pids(),
&[0x0064, 0x0065, 0x0066],
"ca_pids must be the dedup+sorted union of both services' ca_pids"
);
m.set_cat(&[CatCaEntry {
ca_system_id: 0x0648,
ca_pid: 0x1FF0,
private_data: Vec::new(),
}]);
m.set_cam_caids([0x0648].into_iter().collect());
assert_eq!(m.emm_pids(), &[0x1FF0], "precondition: emm_pids populated");
assert_eq!(
m.required_pids(),
vec![0x0064, 0x0065, 0x0066, 0x0100, 0x0101, 0x0200, 0x1FF0],
"required_pids must be descramble_pids ∪ ca_pids ∪ emm_pids"
);
}
#[test]
fn service_of_captures_pcr_pid_and_required_pids_includes_dedicated_pcr() {
use broadcast_common::Parse;
use dvb_si::tables::pmt::PmtSection;
let pmt_bytes = crate::driver::tests::build_ca_pmt_fixture_dedicated_pcr(1550);
let pmt = PmtSection::parse(&pmt_bytes).unwrap();
assert_eq!(
pmt.pcr_pid, 0x00FF,
"fixture precondition: dedicated PCR PID outside the ES/CA set"
);
let svc = service_of(&pmt, CaPmtCmdId::OkDescrambling, vec![], vec![]);
assert_eq!(svc.pcr_pid, 0x00FF, "service_of must capture pmt.pcr_pid");
assert_eq!(svc.es_pids, vec![0x0100, 0x0101]);
assert_eq!(svc.ca_pids, vec![0x0064, 0x0065]);
let mut m = ManagedCa::new();
m.record(1550, svc);
assert!(
m.required_pids().contains(&0x00FF),
"required_pids must include the service's dedicated PCR PID, got {:?}",
m.required_pids()
);
}
#[test]
fn required_pids_excludes_pcr_pid_none() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x0100],
ca_pids: vec![0x0064],
pcr_pid: PCR_PID_NONE,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert_eq!(
m.required_pids(),
vec![0x0064, 0x0100],
"PCR_PID_NONE (0x1FFF) must never be added to required_pids"
);
}
#[test]
fn required_pids_pcr_pid_matching_an_es_pid_adds_no_spurious_pid() {
let mut m = ManagedCa::new();
m.record(
1,
ManagedService {
es_pids: vec![0x0100],
ca_pids: vec![0x0064],
pcr_pid: 0x0100,
cmd: CaPmtCmdId::OkDescrambling,
last_ca_enable: None,
last_descrambling_ok: false,
built_ca_pmt: vec![],
pmt_raw: vec![],
},
);
assert_eq!(
m.required_pids(),
vec![0x0064, 0x0100],
"a pcr_pid coinciding with an ES PID must not duplicate/add a spurious entry"
);
}
}