use std::path::{Path, PathBuf};
use std::{error::Error, fmt};

use clap::ValueEnum;
use serde::{Deserialize, Serialize};

use super::keystore::write_keystore_bytes_atomically;
use super::{owner_id_from_verifying_key, CryptoError, OwnerKeypair};

pub const NODE_OWNERSHIP_VERSION: u32 = 1;
pub const TRUST_STORE_VERSION: u32 = 1;
pub const DEFAULT_NODE_CERT_LIFETIME_SECS: u64 = 7 * 24 * 60 * 60;
pub const DEFAULT_NODE_CERT_RENEW_WINDOW_SECS: u64 = 36 * 60 * 60;
const SIGNING_DOMAIN_TAG: &[u8] = b"mesh-llm-node-ownership-v1:";

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, ValueEnum, Default)]
#[serde(rename_all = "kebab-case")]
pub enum TrustPolicy {
    #[default]
    Off,
    PreferOwned,
    RequireOwned,
    Allowlist,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct NodeOwnershipClaim {
    pub version: u32,
    pub cert_id: String,
    pub owner_id: String,
    pub owner_sign_public_key: String,
    pub node_endpoint_id: String,
    pub issued_at_unix_ms: u64,
    pub expires_at_unix_ms: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub node_label: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hostname_hint: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SignedNodeOwnership {
    pub claim: NodeOwnershipClaim,
    pub signature: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TrustedOwner {
    pub owner_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub label: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RevokedOwner {
    pub owner_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RevokedNodeCert {
    pub cert_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RevokedNodeId {
    pub node_endpoint_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TrustStore {
    pub version: u32,
    pub policy: TrustPolicy,
    #[serde(default)]
    pub trusted_owners: Vec<TrustedOwner>,
    #[serde(default)]
    pub revoked_owners: Vec<RevokedOwner>,
    #[serde(default)]
    pub revoked_node_certs: Vec<RevokedNodeCert>,
    #[serde(default)]
    pub revoked_node_ids: Vec<RevokedNodeId>,
}

impl Default for TrustStore {
    fn default() -> Self {
        Self {
            version: TRUST_STORE_VERSION,
            policy: TrustPolicy::Off,
            trusted_owners: Vec::new(),
            revoked_owners: Vec::new(),
            revoked_node_certs: Vec::new(),
            revoked_node_ids: Vec::new(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum OwnershipStatus {
    Verified,
    #[default]
    Unsigned,
    Expired,
    InvalidSignature,
    MismatchedNodeId,
    RevokedOwner,
    RevokedCert,
    RevokedNodeId,
    UnsupportedProtocol,
    UntrustedOwner,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
pub struct OwnershipSummary {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub owner_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cert_id: Option<String>,
    pub status: OwnershipStatus,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub expires_at_unix_ms: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub node_label: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hostname_hint: Option<String>,
    pub verified: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ControlPlaneAuthError {
    MissingLocalOwnerIdentity {
        local_status: OwnershipStatus,
    },
    MissingRemoteOwnerAttestation,
    RemoteOwnerMismatch {
        local_owner_id: String,
        remote_owner_id: String,
    },
    RemoteOwnershipInvalid {
        status: OwnershipStatus,
        owner_id: Option<String>,
        cert_id: Option<String>,
    },
    TargetNodeMismatch {
        expected_node_id: String,
        actual_node_id: String,
    },
    UnsupportedTrustPolicy {
        policy: TrustPolicy,
    },
}

impl fmt::Display for ControlPlaneAuthError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::MissingLocalOwnerIdentity { local_status } => {
                write!(f, "missing local owner identity ({local_status:?})")
            }
            Self::MissingRemoteOwnerAttestation => write!(f, "missing remote owner attestation"),
            Self::RemoteOwnerMismatch {
                local_owner_id,
                remote_owner_id,
            } => write!(
                f,
                "remote owner mismatch (local {local_owner_id}, remote {remote_owner_id})"
            ),
            Self::RemoteOwnershipInvalid {
                status,
                owner_id,
                cert_id,
            } => write!(
                f,
                "remote ownership invalid ({status:?}, owner_id={}, cert_id={})",
                owner_id.as_deref().unwrap_or("unknown"),
                cert_id.as_deref().unwrap_or("unknown")
            ),
            Self::TargetNodeMismatch {
                expected_node_id,
                actual_node_id,
            } => write!(
                f,
                "target node mismatch (expected {expected_node_id}, got {actual_node_id})"
            ),
            Self::UnsupportedTrustPolicy { policy } => {
                write!(f, "unsupported control-plane trust policy {policy:?}")
            }
        }
    }
}

impl Error for ControlPlaneAuthError {}

fn mesh_dir() -> Result<PathBuf, CryptoError> {
    let home = dirs::home_dir().ok_or_else(|| {
        CryptoError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            "cannot determine home directory",
        ))
    })?;
    Ok(home.join(".mesh-llm"))
}

pub fn default_node_ownership_path() -> Result<PathBuf, CryptoError> {
    Ok(mesh_dir()?.join("node-ownership.json"))
}

pub fn default_trust_store_path() -> Result<PathBuf, CryptoError> {
    Ok(mesh_dir()?.join("trusted-owners.json"))
}

pub fn load_node_ownership(path: &Path) -> Result<SignedNodeOwnership, CryptoError> {
    let raw = std::fs::read_to_string(path)?;
    serde_json::from_str(&raw).map_err(Into::into)
}

pub fn save_node_ownership(
    path: &Path,
    ownership: &SignedNodeOwnership,
) -> Result<(), CryptoError> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let bytes = serde_json::to_vec_pretty(ownership)?;
    write_keystore_bytes_atomically(path, &bytes)?;
    Ok(())
}

pub fn load_trust_store(path: &Path) -> Result<TrustStore, CryptoError> {
    if !path.exists() {
        return Ok(TrustStore::default());
    }
    let raw = std::fs::read_to_string(path)?;
    let store: TrustStore = serde_json::from_str(&raw)?;
    if store.version != TRUST_STORE_VERSION {
        return Err(CryptoError::UnsupportedVersion {
            version: store.version,
        });
    }
    Ok(store)
}

pub fn save_trust_store(path: &Path, store: &TrustStore) -> Result<(), CryptoError> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let bytes = serde_json::to_vec_pretty(store)?;
    write_keystore_bytes_atomically(path, &bytes)?;
    Ok(())
}

fn current_time_unix_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

fn decode_hex_32(label: &str, value: &str) -> Result<[u8; 32], CryptoError> {
    let decoded = hex::decode(value).map_err(|e| CryptoError::InvalidKeyMaterial {
        reason: format!("bad {label} hex: {e}"),
    })?;
    decoded
        .try_into()
        .map_err(|_| CryptoError::InvalidKeyMaterial {
            reason: format!("{label} must be 32 bytes"),
        })
}

fn write_string(buf: &mut Vec<u8>, value: &str) {
    buf.extend_from_slice(&(value.len() as u64).to_le_bytes());
    buf.extend_from_slice(value.as_bytes());
}

fn write_optional_string(buf: &mut Vec<u8>, value: Option<&str>) {
    match value {
        Some(value) => {
            buf.push(1);
            write_string(buf, value);
        }
        None => buf.push(0),
    }
}

fn canonical_claim_bytes(claim: &NodeOwnershipClaim) -> Result<Vec<u8>, CryptoError> {
    let owner_sign_public_key =
        decode_hex_32("owner_sign_public_key", &claim.owner_sign_public_key)?;
    let node_endpoint_id = decode_hex_32("node_endpoint_id", &claim.node_endpoint_id)?;
    let mut buf = Vec::with_capacity(256);
    buf.extend_from_slice(SIGNING_DOMAIN_TAG);
    buf.extend_from_slice(&claim.version.to_le_bytes());
    write_string(&mut buf, &claim.cert_id);
    write_string(&mut buf, &claim.owner_id);
    buf.extend_from_slice(&owner_sign_public_key);
    buf.extend_from_slice(&node_endpoint_id);
    buf.extend_from_slice(&claim.issued_at_unix_ms.to_le_bytes());
    buf.extend_from_slice(&claim.expires_at_unix_ms.to_le_bytes());
    write_optional_string(&mut buf, claim.node_label.as_deref());
    write_optional_string(&mut buf, claim.hostname_hint.as_deref());
    Ok(buf)
}

fn random_cert_id() -> String {
    let bytes: [u8; 16] = rand::random();
    hex::encode(bytes)
}

pub fn sign_node_ownership(
    owner: &OwnerKeypair,
    node_endpoint_id: &[u8; 32],
    expires_at_unix_ms: u64,
    node_label: Option<String>,
    hostname_hint: Option<String>,
) -> Result<SignedNodeOwnership, CryptoError> {
    let claim = NodeOwnershipClaim {
        version: NODE_OWNERSHIP_VERSION,
        cert_id: random_cert_id(),
        owner_id: owner.owner_id(),
        owner_sign_public_key: hex::encode(owner.verifying_key().as_bytes()),
        node_endpoint_id: hex::encode(node_endpoint_id),
        issued_at_unix_ms: current_time_unix_ms(),
        expires_at_unix_ms,
        node_label,
        hostname_hint,
    };
    let bytes = canonical_claim_bytes(&claim)?;
    let signature = owner.sign_bytes(&bytes);
    Ok(SignedNodeOwnership {
        claim,
        signature: hex::encode(signature),
    })
}

pub fn certificate_needs_renewal(
    ownership: &SignedNodeOwnership,
    renew_window_secs: u64,
    now_unix_ms: u64,
) -> bool {
    if ownership.claim.expires_at_unix_ms <= now_unix_ms {
        return true;
    }
    let remaining_ms = ownership
        .claim
        .expires_at_unix_ms
        .saturating_sub(now_unix_ms);
    remaining_ms <= renew_window_secs.saturating_mul(1000)
}

pub fn verify_node_ownership(
    ownership: Option<&SignedNodeOwnership>,
    actual_node_endpoint_id: &[u8; 32],
    trust_store: &TrustStore,
    policy: TrustPolicy,
    now_unix_ms: u64,
) -> OwnershipSummary {
    let Some(ownership) = ownership else {
        return OwnershipSummary::default();
    };

    let mut summary = OwnershipSummary {
        owner_id: Some(ownership.claim.owner_id.clone()),
        cert_id: Some(ownership.claim.cert_id.clone()),
        expires_at_unix_ms: Some(ownership.claim.expires_at_unix_ms),
        node_label: ownership.claim.node_label.clone(),
        hostname_hint: ownership.claim.hostname_hint.clone(),
        ..OwnershipSummary::default()
    };
    if ownership.claim.version != NODE_OWNERSHIP_VERSION {
        summary.status = OwnershipStatus::UnsupportedProtocol;
        return summary;
    }

    let owner_sign_public_key = match decode_hex_32(
        "owner_sign_public_key",
        &ownership.claim.owner_sign_public_key,
    ) {
        Ok(bytes) => bytes,
        Err(_) => {
            summary.status = OwnershipStatus::InvalidSignature;
            return summary;
        }
    };
    let node_endpoint_id =
        match decode_hex_32("node_endpoint_id", &ownership.claim.node_endpoint_id) {
            Ok(bytes) => bytes,
            Err(_) => {
                summary.status = OwnershipStatus::InvalidSignature;
                return summary;
            }
        };

    if node_endpoint_id != *actual_node_endpoint_id {
        summary.status = OwnershipStatus::MismatchedNodeId;
        return summary;
    }

    let verifying_key = match ed25519_dalek::VerifyingKey::from_bytes(&owner_sign_public_key) {
        Ok(value) => value,
        Err(_) => {
            summary.status = OwnershipStatus::InvalidSignature;
            return summary;
        }
    };
    let expected_owner_id = owner_id_from_verifying_key(&verifying_key);
    if expected_owner_id != ownership.claim.owner_id {
        summary.status = OwnershipStatus::InvalidSignature;
        return summary;
    }

    let canonical = match canonical_claim_bytes(&ownership.claim) {
        Ok(value) => value,
        Err(_) => {
            summary.status = OwnershipStatus::InvalidSignature;
            return summary;
        }
    };
    let sig_bytes = match hex::decode(&ownership.signature) {
        Ok(value) => value,
        Err(_) => {
            summary.status = OwnershipStatus::InvalidSignature;
            return summary;
        }
    };
    let sig_bytes: [u8; 64] = match sig_bytes.try_into() {
        Ok(value) => value,
        Err(_) => {
            summary.status = OwnershipStatus::InvalidSignature;
            return summary;
        }
    };
    let signature = ed25519_dalek::Signature::from_bytes(&sig_bytes);
    if verifying_key.verify_strict(&canonical, &signature).is_err() {
        summary.status = OwnershipStatus::InvalidSignature;
        return summary;
    }

    if ownership.claim.expires_at_unix_ms <= now_unix_ms {
        summary.status = OwnershipStatus::Expired;
        return summary;
    }

    if trust_store
        .revoked_owners
        .iter()
        .any(|entry| entry.owner_id == ownership.claim.owner_id)
    {
        summary.status = OwnershipStatus::RevokedOwner;
        return summary;
    }

    if trust_store
        .revoked_node_certs
        .iter()
        .any(|entry| entry.cert_id == ownership.claim.cert_id)
    {
        summary.status = OwnershipStatus::RevokedCert;
        return summary;
    }

    if trust_store.revoked_node_ids.iter().any(|entry| {
        entry
            .node_endpoint_id
            .eq_ignore_ascii_case(&ownership.claim.node_endpoint_id)
    }) {
        summary.status = OwnershipStatus::RevokedNodeId;
        return summary;
    }

    if matches!(policy, TrustPolicy::Allowlist)
        && !trust_store
            .trusted_owners
            .iter()
            .any(|entry| entry.owner_id == ownership.claim.owner_id)
    {
        summary.status = OwnershipStatus::UntrustedOwner;
        return summary;
    }

    summary.status = OwnershipStatus::Verified;
    summary.verified = true;
    summary
}

pub fn verify_control_plane_target_node(
    target_node_id: &[u8],
    actual_local_endpoint_id: &[u8; 32],
) -> Result<(), ControlPlaneAuthError> {
    if target_node_id == actual_local_endpoint_id {
        return Ok(());
    }
    Err(ControlPlaneAuthError::TargetNodeMismatch {
        expected_node_id: hex::encode(actual_local_endpoint_id),
        actual_node_id: hex::encode(target_node_id),
    })
}

pub fn verify_control_plane_peer_ownership(
    local_owner: &OwnershipSummary,
    remote_ownership: Option<&crate::proto::node::SignedNodeOwnership>,
    actual_remote_endpoint_id: &[u8; 32],
    trust_store: &TrustStore,
    trust_policy: TrustPolicy,
    now_unix_ms: u64,
) -> Result<OwnershipSummary, ControlPlaneAuthError> {
    let Some(local_owner_id) = local_owner
        .owner_id
        .as_ref()
        .filter(|_| local_owner.verified)
    else {
        return Err(ControlPlaneAuthError::MissingLocalOwnerIdentity {
            local_status: local_owner.status.clone(),
        });
    };

    match trust_policy {
        TrustPolicy::Off | TrustPolicy::PreferOwned | TrustPolicy::RequireOwned => {}
        TrustPolicy::Allowlist => {
            return Err(ControlPlaneAuthError::UnsupportedTrustPolicy {
                policy: trust_policy,
            });
        }
    }

    let remote_ownership = remote_ownership
        .map(proto_signed_node_ownership_to_local)
        .ok_or(ControlPlaneAuthError::MissingRemoteOwnerAttestation)?;
    let remote_summary = verify_node_ownership(
        Some(&remote_ownership),
        actual_remote_endpoint_id,
        trust_store,
        TrustPolicy::Off,
        now_unix_ms,
    );

    match remote_summary.status {
        OwnershipStatus::Verified => {
            let Some(remote_owner_id) = remote_summary.owner_id.as_ref() else {
                return Err(ControlPlaneAuthError::RemoteOwnershipInvalid {
                    status: remote_summary.status.clone(),
                    owner_id: None,
                    cert_id: remote_summary.cert_id.clone(),
                });
            };
            if remote_owner_id != local_owner_id {
                return Err(ControlPlaneAuthError::RemoteOwnerMismatch {
                    local_owner_id: local_owner_id.clone(),
                    remote_owner_id: remote_owner_id.clone(),
                });
            }
            Ok(remote_summary)
        }
        OwnershipStatus::MismatchedNodeId => Err(ControlPlaneAuthError::TargetNodeMismatch {
            expected_node_id: hex::encode(actual_remote_endpoint_id),
            actual_node_id: remote_ownership.claim.node_endpoint_id,
        }),
        _ => Err(ControlPlaneAuthError::RemoteOwnershipInvalid {
            status: remote_summary.status.clone(),
            owner_id: remote_summary.owner_id.clone(),
            cert_id: remote_summary.cert_id.clone(),
        }),
    }
}

fn proto_signed_node_ownership_to_local(
    attestation: &crate::proto::node::SignedNodeOwnership,
) -> SignedNodeOwnership {
    SignedNodeOwnership {
        claim: NodeOwnershipClaim {
            version: attestation.version,
            cert_id: attestation.cert_id.clone(),
            owner_id: attestation.owner_id.clone(),
            owner_sign_public_key: hex::encode(&attestation.owner_sign_public_key),
            node_endpoint_id: hex::encode(&attestation.node_endpoint_id),
            issued_at_unix_ms: attestation.issued_at_unix_ms,
            expires_at_unix_ms: attestation.expires_at_unix_ms,
            node_label: attestation.node_label.clone(),
            hostname_hint: attestation.hostname_hint.clone(),
        },
        signature: hex::encode(&attestation.signature),
    }
}

impl TrustStore {
    pub fn merged_with_trusted_owners(mut self, owners: &[String]) -> Self {
        for owner_id in owners {
            self.add_trusted_owner(owner_id.clone(), None);
        }
        self
    }

    pub fn add_trusted_owner(&mut self, owner_id: String, label: Option<String>) {
        if let Some(existing) = self
            .trusted_owners
            .iter_mut()
            .find(|entry| entry.owner_id == owner_id)
        {
            if label.is_some() {
                existing.label = label;
            }
            return;
        }
        self.trusted_owners.push(TrustedOwner { owner_id, label });
        self.trusted_owners
            .sort_by(|a, b| a.owner_id.cmp(&b.owner_id));
    }

    pub fn remove_trusted_owner(&mut self, owner_id: &str) -> bool {
        let before = self.trusted_owners.len();
        self.trusted_owners
            .retain(|entry| entry.owner_id != owner_id);
        before != self.trusted_owners.len()
    }

    pub fn revoke_owner(&mut self, owner_id: String, reason: Option<String>) {
        if self
            .revoked_owners
            .iter()
            .any(|entry| entry.owner_id == owner_id)
        {
            return;
        }
        self.revoked_owners.push(RevokedOwner { owner_id, reason });
        self.revoked_owners
            .sort_by(|a, b| a.owner_id.cmp(&b.owner_id));
    }

    pub fn revoke_node_cert(&mut self, cert_id: String, reason: Option<String>) {
        if self
            .revoked_node_certs
            .iter()
            .any(|entry| entry.cert_id == cert_id)
        {
            return;
        }
        self.revoked_node_certs
            .push(RevokedNodeCert { cert_id, reason });
        self.revoked_node_certs
            .sort_by(|a, b| a.cert_id.cmp(&b.cert_id));
    }

    pub fn revoke_node_id(&mut self, node_endpoint_id: String, reason: Option<String>) {
        if self.revoked_node_ids.iter().any(|entry| {
            entry
                .node_endpoint_id
                .eq_ignore_ascii_case(&node_endpoint_id)
        }) {
            return;
        }
        self.revoked_node_ids.push(RevokedNodeId {
            node_endpoint_id,
            reason,
        });
        self.revoked_node_ids
            .sort_by(|a, b| a.node_endpoint_id.cmp(&b.node_endpoint_id));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn proto_signed_node_ownership(
        ownership: &SignedNodeOwnership,
    ) -> crate::proto::node::SignedNodeOwnership {
        crate::proto::node::SignedNodeOwnership {
            version: ownership.claim.version,
            cert_id: ownership.claim.cert_id.clone(),
            owner_id: ownership.claim.owner_id.clone(),
            owner_sign_public_key: hex::decode(&ownership.claim.owner_sign_public_key)
                .expect("test owner_sign_public_key must decode"),
            node_endpoint_id: hex::decode(&ownership.claim.node_endpoint_id)
                .expect("test node_endpoint_id must decode"),
            issued_at_unix_ms: ownership.claim.issued_at_unix_ms,
            expires_at_unix_ms: ownership.claim.expires_at_unix_ms,
            node_label: ownership.claim.node_label.clone(),
            hostname_hint: ownership.claim.hostname_hint.clone(),
            signature: hex::decode(&ownership.signature).expect("test signature must decode"),
        }
    }

    fn verified_local_owner_summary(owner: &OwnerKeypair) -> OwnershipSummary {
        OwnershipSummary {
            owner_id: Some(owner.owner_id()),
            status: OwnershipStatus::Verified,
            verified: true,
            ..OwnershipSummary::default()
        }
    }

    #[test]
    fn signed_ownership_verifies() {
        let owner = OwnerKeypair::generate();
        let node_endpoint_id = [0x42; 32];
        let ownership = sign_node_ownership(
            &owner,
            &node_endpoint_id,
            current_time_unix_ms() + 60_000,
            Some("studio".into()),
            Some("studio-host".into()),
        )
        .unwrap();

        let summary = verify_node_ownership(
            Some(&ownership),
            &node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        );

        assert_eq!(summary.status, OwnershipStatus::Verified);
        assert!(summary.verified);
        assert_eq!(summary.node_label.as_deref(), Some("studio"));
    }

    #[test]
    fn mismatched_node_id_is_rejected() {
        let owner = OwnerKeypair::generate();
        let node_endpoint_id = [0x42; 32];
        let wrong_node_endpoint_id = [0x24; 32];
        let ownership = sign_node_ownership(
            &owner,
            &node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let summary = verify_node_ownership(
            Some(&ownership),
            &wrong_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        );

        assert_eq!(summary.status, OwnershipStatus::MismatchedNodeId);
        assert!(!summary.verified);
    }

    #[test]
    fn revoked_owner_is_rejected() {
        let owner = OwnerKeypair::generate();
        let node_endpoint_id = [0x42; 32];
        let ownership = sign_node_ownership(
            &owner,
            &node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();
        let mut trust_store = TrustStore::default();
        trust_store.revoke_owner(owner.owner_id(), Some("compromised".into()));

        let summary = verify_node_ownership(
            Some(&ownership),
            &node_endpoint_id,
            &trust_store,
            TrustPolicy::Off,
            current_time_unix_ms(),
        );

        assert_eq!(summary.status, OwnershipStatus::RevokedOwner);
    }

    #[test]
    fn allowlist_requires_trusted_owner() {
        let owner = OwnerKeypair::generate();
        let node_endpoint_id = [0x42; 32];
        let ownership = sign_node_ownership(
            &owner,
            &node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let summary = verify_node_ownership(
            Some(&ownership),
            &node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Allowlist,
            current_time_unix_ms(),
        );

        assert_eq!(summary.status, OwnershipStatus::UntrustedOwner);
    }

    #[test]
    fn unsupported_claim_version_is_rejected() {
        let owner = OwnerKeypair::generate();
        let node_endpoint_id = [0x42; 32];
        let mut ownership = sign_node_ownership(
            &owner,
            &node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();
        ownership.claim.version = NODE_OWNERSHIP_VERSION + 1;

        let summary = verify_node_ownership(
            Some(&ownership),
            &node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        );

        assert_eq!(summary.status, OwnershipStatus::UnsupportedProtocol);
        assert!(!summary.verified);
    }

    #[test]
    fn control_plane_auth_same_owner_without_gossip() {
        let owner = OwnerKeypair::generate();
        let local_owner = verified_local_owner_summary(&owner);
        let remote_node_endpoint_id = [0x52; 32];
        let remote_ownership = sign_node_ownership(
            &owner,
            &remote_node_endpoint_id,
            current_time_unix_ms() + 60_000,
            Some("remote-worker".into()),
            None,
        )
        .unwrap();

        let summary = verify_control_plane_peer_ownership(
            &local_owner,
            Some(&proto_signed_node_ownership(&remote_ownership)),
            &remote_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        )
        .expect("same-owner direct control attestation must succeed without gossip state");

        assert!(summary.verified);
        assert_eq!(summary.owner_id.as_deref(), Some(owner.owner_id().as_str()));
        assert_eq!(summary.node_label.as_deref(), Some("remote-worker"));
    }

    #[test]
    fn control_plane_auth_rejects_wrong_owner() {
        let local_owner = OwnerKeypair::generate();
        let remote_owner = OwnerKeypair::generate();
        let remote_node_endpoint_id = [0x62; 32];
        let remote_ownership = sign_node_ownership(
            &remote_owner,
            &remote_node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let err = verify_control_plane_peer_ownership(
            &verified_local_owner_summary(&local_owner),
            Some(&proto_signed_node_ownership(&remote_ownership)),
            &remote_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        )
        .expect_err("different-owner control attestation must fail closed");

        assert!(matches!(
            err,
            ControlPlaneAuthError::RemoteOwnerMismatch { .. }
        ));
    }

    #[test]
    fn control_plane_auth_rejects_wrong_node_id() {
        let owner = OwnerKeypair::generate();
        let local_owner = verified_local_owner_summary(&owner);
        let claimed_node_endpoint_id = [0x71; 32];
        let actual_remote_endpoint_id = [0x72; 32];
        let remote_ownership = sign_node_ownership(
            &owner,
            &claimed_node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let err = verify_control_plane_peer_ownership(
            &local_owner,
            Some(&proto_signed_node_ownership(&remote_ownership)),
            &actual_remote_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        )
        .expect_err("wrong peer node id must fail closed");

        assert!(matches!(
            err,
            ControlPlaneAuthError::TargetNodeMismatch { .. }
        ));
    }

    #[test]
    fn control_plane_auth_rejects_bad_signature() {
        let owner = OwnerKeypair::generate();
        let local_owner = verified_local_owner_summary(&owner);
        let remote_node_endpoint_id = [0x81; 32];
        let mut remote_ownership = proto_signed_node_ownership(
            &sign_node_ownership(
                &owner,
                &remote_node_endpoint_id,
                current_time_unix_ms() + 60_000,
                None,
                None,
            )
            .unwrap(),
        );
        remote_ownership.signature[0] ^= 0xFF;

        let err = verify_control_plane_peer_ownership(
            &local_owner,
            Some(&remote_ownership),
            &remote_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        )
        .expect_err("bad-signature control attestation must fail closed");

        assert!(matches!(
            err,
            ControlPlaneAuthError::RemoteOwnershipInvalid {
                status: OwnershipStatus::InvalidSignature,
                ..
            }
        ));
    }

    #[test]
    fn control_plane_auth_rejects_missing_local_owner_identity() {
        let owner = OwnerKeypair::generate();
        let remote_node_endpoint_id = [0x91; 32];
        let remote_ownership = sign_node_ownership(
            &owner,
            &remote_node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let err = verify_control_plane_peer_ownership(
            &OwnershipSummary::default(),
            Some(&proto_signed_node_ownership(&remote_ownership)),
            &remote_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Off,
            current_time_unix_ms(),
        )
        .expect_err("missing local owner identity must fail closed");

        assert!(matches!(
            err,
            ControlPlaneAuthError::MissingLocalOwnerIdentity {
                local_status: OwnershipStatus::Unsigned,
            }
        ));
    }

    #[test]
    fn control_plane_auth_rejects_unsupported_trust_policy() {
        let owner = OwnerKeypair::generate();
        let local_owner = verified_local_owner_summary(&owner);
        let remote_node_endpoint_id = [0xA1; 32];
        let remote_ownership = sign_node_ownership(
            &owner,
            &remote_node_endpoint_id,
            current_time_unix_ms() + 60_000,
            None,
            None,
        )
        .unwrap();

        let err = verify_control_plane_peer_ownership(
            &local_owner,
            Some(&proto_signed_node_ownership(&remote_ownership)),
            &remote_node_endpoint_id,
            &TrustStore::default(),
            TrustPolicy::Allowlist,
            current_time_unix_ms(),
        )
        .expect_err("allowlist trust policy must fail closed for owner-control auth");

        assert!(matches!(
            err,
            ControlPlaneAuthError::UnsupportedTrustPolicy {
                policy: TrustPolicy::Allowlist,
            }
        ));
    }

    #[test]
    fn control_plane_auth_rejects_target_node_mismatch() {
        let err = verify_control_plane_target_node(&[0xCD; 32], &[0xAB; 32])
            .expect_err("wrong target node id must fail closed");

        assert!(matches!(
            err,
            ControlPlaneAuthError::TargetNodeMismatch { .. }
        ));
    }
}
