bacnet-server 0.12.0

//! Original accepted connection replies; fallback observations are not proof
//! of delivery to another peer. Every ingress envelope comes from real TLS.
use super::*;
use bacnet_services::read_property::{ReadPropertyACK, ReadPropertyRequest};
use bacnet_transport::sc_frame::ScFunction;

fn read_name(invoke: u8) -> Apdu {
    let mut data = BytesMut::new();
    ReadPropertyRequest {
        object_identifier: csv_oid(),
        property_identifier: PropertyIdentifier::OBJECT_NAME,
        property_array_index: None,
    }
    .encode(&mut data);
    confirmed(ConfirmedServiceChoice::READ_PROPERTY, invoke, data.freeze())
}

async fn socket_response(f: &mut Fixture, peer: &Peer) -> Apdu {
    let wire = bounded(async {
        tokio::select! {
            reply = peer.ws.recv() => reply.unwrap(),
            fallback = f.responses.recv() => panic!("direct reply used address-based fallback: {fallback:?}"),
        }
    })
    .await;
    let frame = decode_sc_message(&wire).unwrap();
    assert_eq!(frame.function, ScFunction::EncapsulatedNpdu);
    assert_eq!(frame.originating_vmac, None);
    assert_eq!(frame.destination_vmac, None);
    let npdu = decode_npdu(frame.payload).unwrap();
    let destination = npdu.destination.unwrap();
    assert_eq!(destination.network, 123);
    assert_eq!(destination.mac_address.as_ref(), &[3]);
    apdu::decode_apdu(npdu.payload).unwrap()
}

fn assert_name(response: Apdu, invoke: u8) {
    let Apdu::ComplexAck(ack) = response else {
        panic!("expected data-bearing ReadProperty response");
    };
    assert_eq!(ack.invoke_id, invoke);
    let ack = ReadPropertyACK::decode(&ack.service_ack).unwrap();
    assert_eq!(ack.object_identifier, csv_oid());
    assert_eq!(ack.property_identifier, PropertyIdentifier::OBJECT_NAME);
    let mut expected = BytesMut::new();
    encode_property_value(
        &mut expected,
        &PropertyValue::CharacterString("value".into()),
    )
    .unwrap();
    assert_eq!(ack.property_value, expected.to_vec());
}

#[tokio::test]
async fn direct_response_same_socket_read_property() {
    let ca = TestCa::new();
    let mut f = Fixture::new(
        &ca,
        ServerConfig::default(),
        database(Arc::new(AtomicUsize::new(0))),
    )
    .await;
    let mut a = f.peer(ca.tls("a")).await;
    let envelope = f.capture(&mut a, &read_name(41)).await;
    f.feed(envelope).await;
    assert_name(socket_response(&mut f, &a).await, 41);
    f.active(0).await;
    assert!(f.responses.try_recv().is_err());
    f.stop().await;
}

#[tokio::test]
async fn direct_response_held_read_replacement_has_no_address_fallback() {
    let ca = TestCa::new();
    let mut f = Fixture::new(
        &ca,
        ServerConfig::default(),
        database(Arc::new(AtomicUsize::new(0))),
    )
    .await;
    let mut a = f.peer(ca.tls("a")).await;
    let envelope = f.capture(&mut a, &read_name(41)).await;
    let identity_a = envelope.provenance.direct_sc_identity().unwrap();
    let db = f.server.db.clone();
    let held = db.write().await;
    f.feed(envelope).await;
    f.active(1).await;

    let mut b = f.peer(ca.tls("b")).await;
    let b_request = f.capture(&mut b, &read_name(42)).await;
    let identity_b = b_request.provenance.direct_sc_identity().unwrap();
    assert_ne!(identity_a.leaf_sha256(), identity_b.leaf_sha256());
    assert_ne!(identity_a.incarnation(), identity_b.incarnation());
    drop(held);
    f.active(0).await;
    assert!(
        f.responses.try_recv().is_err(),
        "retired A reply reached address-based egress"
    );

    // FIFO socket observation after A's task completes: B's first application
    // frame must answer B, never the held A transaction.
    f.feed(b_request).await;
    assert_name(socket_response(&mut f, &b).await, 42);
    f.stop().await;
}

#[tokio::test]
async fn direct_response_success_error_reject_and_oversize_abort_use_original_socket() {
    let ca = TestCa::new();
    let mut f = Fixture::new(
        &ca,
        ServerConfig::default(),
        database(Arc::new(AtomicUsize::new(0))),
    )
    .await;
    let mut a = f.peer(ca.tls("a")).await;
    let write = confirmed(
        ConfirmedServiceChoice::WRITE_PROPERTY,
        1,
        write_payload(&"x".repeat(200)),
    );
    let envelope = f.capture(&mut a, &write).await;
    f.feed(envelope).await;
    assert!(
        matches!(socket_response(&mut f, &a).await, Apdu::SimpleAck(ack) if ack.invoke_id == 1)
    );

    let mut missing = BytesMut::new();
    ReadPropertyRequest {
        object_identifier: csv_oid(),
        property_identifier: PropertyIdentifier::from_raw(5555),
        property_array_index: None,
    }
    .encode(&mut missing);
    let invalid = confirmed(ConfirmedServiceChoice::READ_PROPERTY, 2, missing.freeze());
    let envelope = f.capture(&mut a, &invalid).await;
    f.feed(envelope).await;
    assert!(
        matches!(socket_response(&mut f, &a).await, Apdu::Error(error) if error.invoke_id == 2 && error.error_class == ErrorClass::PROPERTY && error.error_code == ErrorCode::UNKNOWN_PROPERTY)
    );
    let unknown = confirmed(ConfirmedServiceChoice::from_raw(254), 3, Bytes::new());
    let envelope = f.capture(&mut a, &unknown).await;
    f.feed(envelope).await;
    assert!(matches!(socket_response(&mut f, &a).await, Apdu::Reject(ack) if ack.invoke_id == 3));

    let mut encoded = BytesMut::new();
    ReadPropertyRequest {
        object_identifier: csv_oid(),
        property_identifier: PropertyIdentifier::PRESENT_VALUE,
        property_array_index: None,
    }
    .encode(&mut encoded);
    let Apdu::ConfirmedRequest(mut request) =
        confirmed(ConfirmedServiceChoice::READ_PROPERTY, 4, encoded.freeze())
    else {
        unreachable!()
    };
    request.max_apdu_length = 50;
    let envelope = f.capture(&mut a, &Apdu::ConfirmedRequest(request)).await;
    f.feed(envelope).await;
    assert!(
        matches!(socket_response(&mut f, &a).await, Apdu::Abort(ack) if ack.invoke_id == 4 && ack.abort_reason == AbortReason::SEGMENTATION_NOT_SUPPORTED)
    );
    f.stop().await;
}

#[tokio::test]
async fn direct_response_missing_and_mismatched_capabilities_never_fall_back() {
    let ca = TestCa::new();
    let mut f = Fixture::new(
        &ca,
        ServerConfig::default(),
        database(Arc::new(AtomicUsize::new(0))),
    )
    .await;
    let mut a = f.peer(ca.tls("a")).await;
    let mut missing = f.capture(&mut a, &read_name(1)).await;
    missing.direct_response = None;
    f.feed(missing).await;
    f.dispatch_barrier(&mut a, 240).await;
    f.active(0).await;
    assert!(f.responses.try_recv().is_err());

    let mut old = f.capture(&mut a, &read_name(2)).await;
    let mut b = f.peer(ca.tls("b")).await;
    let new = f.capture(&mut b, &read_name(3)).await;
    old.direct_response = new.direct_response.clone();
    f.feed(old).await;
    f.dispatch_barrier(&mut b, 241).await;
    f.active(0).await;
    assert!(f.responses.try_recv().is_err());
    f.feed(new).await;
    assert_name(socket_response(&mut f, &b).await, 3);
    f.stop().await;
}

#[tokio::test]
async fn direct_response_lso_replay_and_same_leaf_reconnect_keep_original_route() {
    let ca = TestCa::new();
    let tls = ca.tls("a");
    let authorized = Arc::new(AtomicUsize::new(0));
    let observed = authorized.clone();
    let executions = Arc::new(AtomicUsize::new(0));
    let mut f = Fixture::new(
        &ca,
        ServerConfig {
            life_safety_operation_authorizer: Some(Arc::new(move |_| {
                observed.fetch_add(1, Ordering::AcqRel);
                true
            })),
            ..Default::default()
        },
        database(executions.clone()),
    )
    .await;
    let mut a = f.peer(tls.clone()).await;
    let request = request(true, 42);
    let envelope = f.capture(&mut a, &request).await;
    f.feed(envelope).await;
    let first = socket_response(&mut f, &a).await;
    assert!(matches!(first, Apdu::SimpleAck(_)));
    let replay = f.capture(&mut a, &request).await;
    f.feed(replay).await;
    assert_eq!(socket_response(&mut f, &a).await, first);
    assert_eq!(authorized.load(Ordering::Acquire), 1);
    assert_eq!(executions.load(Ordering::Acquire), 1);
    let old_replay = f.capture(&mut a, &request).await;
    let mut b = f.peer(tls).await;
    let new = f.capture(&mut b, &request).await;
    assert_ne!(old_replay.provenance, new.provenance);
    f.feed(old_replay).await;
    f.dispatch_barrier(&mut b, 240).await;
    assert_eq!(authorized.load(Ordering::Acquire), 1);
    assert!(f.responses.try_recv().is_err());
    f.feed(new).await;
    // RESET completed on A and cleared Operation_Expected. A fresh B
    // incarnation reauthorizes and observes current state instead of replaying
    // A's successful ACK.
    assert!(matches!(socket_response(&mut f, &b).await,
        Apdu::Error(error) if error.error_code == ErrorCode::INVALID_OPERATION_IN_THIS_STATE));
    assert_eq!(authorized.load(Ordering::Acquire), 2);
    assert_eq!(executions.load(Ordering::Acquire), 1);
    f.stop().await;
}

#[tokio::test]
async fn direct_response_overload_abort_uses_current_original_socket() {
    let ca = TestCa::new();
    let mut f = Fixture::new(
        &ca,
        ServerConfig {
            request_admission_policy: RequestAdmissionPolicy {
                max_confirmed_in_flight: 1,
                confirmed_recovery_reserve: 0,
                ..Default::default()
            },
            ..Default::default()
        },
        database(Arc::new(AtomicUsize::new(0))),
    )
    .await;
    let mut a = f.peer(ca.tls("a")).await;
    let db = f.server.db.clone();
    let held = db.write().await;
    let first = f.capture(&mut a, &read_name(1)).await;
    f.feed(first).await;
    f.active(1).await;
    let overload = f.capture(&mut a, &read_name(2)).await;
    f.feed(overload).await;
    assert!(
        matches!(socket_response(&mut f, &a).await, Apdu::Abort(a) if a.invoke_id == 2 && a.abort_reason == AbortReason::OUT_OF_RESOURCES)
    );
    drop(held);
    assert_name(socket_response(&mut f, &a).await, 1);
    f.stop().await;
}

#[path = "direct_response_segment_tests.rs"]
mod segments;

#[path = "direct_response_owner_tests.rs"]
mod owners;

#[path = "direct_response_budget_tests.rs"]
mod budgets;