saddle-db 0.2.0-rc.19

//! Strict Database-owned input for a Gate-signed deployment DB observation.
//!
//! Signature, nonce, trusted time and listener authority stay Facade/Gate
//! responsibilities. This module observes the real configured MariaDB service,
//! parses the signed DB member without defaults, and implements the existing
//! linear service-attestation seam. It never serializes a DSN or credential.

use std::{collections::BTreeSet, net::IpAddr, time::Instant};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use sqlx::{
    Connection,
    mysql::{MySqlConnectOptions, MySqlConnection, MySqlSslMode},
};

use crate::{
    DatabaseConfig,
    c1_normal_return::db_normal_return_work_proof,
    c1_service_attestation::{
        DbReturnCalibration, DbReturnServiceDomain, DbReturnServiceRates, DbServiceTarget,
        DeploymentDbServiceAttestation,
    },
};

const SCHEMA: &str = "saddle-deployment-db-observation-v1";
const MAX_OBSERVATION_BYTES: usize = 16 * 1024;
const SQLX_IDENTITY: &[u8] = b"sqlx-mysql/0.8.6;runtime-tokio;tls-rustls-ring-native-roots";
const FORCED_OPTIONS_IDENTITY: &[u8] = b"socket=none;client-cert=none;client-key=none;timezone=none;statement-cache=default;set-names=default;log-settings=default";

/// Sealed identity for the synchronous same-task poison-discard path.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DbPoisonDiscardWorkProof {
    identity: [u8; 32],
}

impl DbPoisonDiscardWorkProof {
    pub const fn identity(self) -> [u8; 32] {
        self.identity
    }
}

pub const fn db_poison_discard_work_proof() -> DbPoisonDiscardWorkProof {
    DbPoisonDiscardWorkProof {
        identity: *b"SDL-DB-DISCARD-1-SQLX-0806-MYSQL",
    }
}

/// Gate-owned identities that cannot be inferred from a DB socket.
#[derive(Clone, Copy, Debug)]
pub struct DbDeploymentObserverIdentity {
    pub environment: [u8; 32],
    pub build: [u8; 32],
    pub gate: [u8; 32],
    pub network_domain: [u8; 32],
    pub credential: [u8; 32],
}

/// Conservative deployment service bounds approved by Gate. These are not
/// calculated from a mean, percentile, or the current observation.
#[derive(Clone, Copy, Debug)]
pub struct ApprovedDbReturnBounds {
    pub ping_command_nanos: u64,
    pub protocol_write_nanos: u64,
    pub protocol_read_nanos: u64,
    pub service_attestation: [u8; 32],
    pub calibration_identity: [u8; 32],
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DbDeploymentObservationError {
    InvalidSchema,
    InvalidIdentity,
    InvalidTls,
    InvalidEndpoint,
    InvalidService,
    MissingApprovedBound,
    ObservationFailed,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Document {
    schema: String,
    release_approval: bool,
    endpoint: Endpoint,
    service: Service,
    proofs: Proofs,
    calibration: Calibration,
    identity: Identity,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Endpoint {
    canonical_hostname: String,
    port: u16,
    database_sha256: [u8; 32],
    username_sha256: [u8; 32],
    credential_identity: [u8; 32],
    tls_mode: String,
    fixed_ca_sha256: [u8; 32],
    connect_options_sha256: [u8; 32],
    observed_dns_set_sha256: [u8; 32],
    network_domain_identity: [u8; 32],
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Service {
    tls_cipher_sha256: [u8; 32],
    tls_version_sha256: [u8; 32],
    mariadb_version_sha256: [u8; 32],
    version_comment_sha256: [u8; 32],
    server_hostname_sha256: [u8; 32],
    server_identity: [u8; 32],
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Proofs {
    normal_return_work_identity: [u8; 32],
    poison_discard_work_identity: [u8; 32],
    sqlx_identity: [u8; 32],
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Calibration {
    approved_by_gate: bool,
    samples: u64,
    max_observed_ping_command_nanos: u64,
    max_observed_protocol_write_nanos: u64,
    max_observed_protocol_read_nanos: u64,
    approved_ping_command_nanos: u64,
    approved_protocol_write_nanos: u64,
    approved_protocol_read_nanos: u64,
    calibration_identity: [u8; 32],
    service_attestation: [u8; 32],
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Identity {
    environment: [u8; 32],
    target_system: [u8; 32],
    build: [u8; 32],
    gate: [u8; 32],
    observation: [u8; 32],
}

/// Concrete non-Clone attestation produced from the strictly parsed signed DB
/// member. It carries no raw JSON, endpoint name, or server status string.
#[derive(Debug)]
pub struct ObservedDeploymentDbAttestation {
    document: Document,
}

/// Parses the already signature-verified canonical DB member. Serde's
/// deny_unknown_fields plus concrete required fields rejects unknown, missing,
/// duplicate and type-drifted input; no default or flatten path exists.
pub fn parse_deployment_db_observation(
    bytes: &[u8],
) -> Result<ObservedDeploymentDbAttestation, DbDeploymentObservationError> {
    if bytes.is_empty() || bytes.len() > MAX_OBSERVATION_BYTES {
        return Err(DbDeploymentObservationError::InvalidSchema);
    }
    let document: Document =
        serde_json::from_slice(bytes).map_err(|_| DbDeploymentObservationError::InvalidSchema)?;
    validate(&document)?;
    if serde_json::to_vec(&document).map_err(|_| DbDeploymentObservationError::InvalidSchema)?
        != bytes
    {
        return Err(DbDeploymentObservationError::InvalidSchema);
    }
    Ok(ObservedDeploymentDbAttestation { document })
}

fn validate(document: &Document) -> Result<(), DbDeploymentObservationError> {
    if document.schema != SCHEMA || document.release_approval {
        return Err(DbDeploymentObservationError::InvalidSchema);
    }
    if document.endpoint.canonical_hostname.is_empty()
        || document.endpoint.canonical_hostname.len() > 253
        || document.endpoint.canonical_hostname
            != document.endpoint.canonical_hostname.to_ascii_lowercase()
        || document
            .endpoint
            .canonical_hostname
            .bytes()
            .any(|byte| byte.is_ascii_whitespace() || byte.is_ascii_control())
        || document.endpoint.port == 0
        || document.endpoint.tls_mode != "verify-identity"
    {
        return Err(DbDeploymentObservationError::InvalidEndpoint);
    }
    let identities = [
        document.endpoint.database_sha256,
        document.endpoint.username_sha256,
        document.endpoint.credential_identity,
        document.endpoint.fixed_ca_sha256,
        document.endpoint.connect_options_sha256,
        document.endpoint.observed_dns_set_sha256,
        document.endpoint.network_domain_identity,
        document.service.tls_cipher_sha256,
        document.service.tls_version_sha256,
        document.service.mariadb_version_sha256,
        document.service.version_comment_sha256,
        document.service.server_hostname_sha256,
        document.service.server_identity,
        document.proofs.normal_return_work_identity,
        document.proofs.poison_discard_work_identity,
        document.proofs.sqlx_identity,
        document.calibration.calibration_identity,
        document.calibration.service_attestation,
        document.identity.environment,
        document.identity.target_system,
        document.identity.build,
        document.identity.gate,
        document.identity.observation,
    ];
    if identities.contains(&[0; 32]) {
        return Err(DbDeploymentObservationError::InvalidIdentity);
    }
    if document.proofs.normal_return_work_identity != db_normal_return_work_proof().identity()
        || document.proofs.poison_discard_work_identity != db_poison_discard_work_proof().identity()
        || document.proofs.sqlx_identity != hash(SQLX_IDENTITY)
    {
        return Err(DbDeploymentObservationError::InvalidIdentity);
    }
    let c = &document.calibration;
    if c.samples == 0
        || c.max_observed_ping_command_nanos == 0
        || c.max_observed_protocol_write_nanos == 0
        || c.max_observed_protocol_read_nanos == 0
    {
        return Err(DbDeploymentObservationError::InvalidService);
    }
    if c.approved_ping_command_nanos == 0
        || c.approved_protocol_write_nanos == 0
        || c.approved_protocol_read_nanos == 0
        || !c.approved_by_gate
    {
        return Err(DbDeploymentObservationError::MissingApprovedBound);
    }
    if c.approved_ping_command_nanos < c.max_observed_ping_command_nanos
        || c.approved_protocol_write_nanos < c.max_observed_protocol_write_nanos
        || c.approved_protocol_read_nanos < c.max_observed_protocol_read_nanos
    {
        return Err(DbDeploymentObservationError::InvalidService);
    }
    Ok(())
}

/// Observes a real TLS-authenticated MariaDB endpoint and emits canonical JSON
/// for external Gate signing. The whole ping round trip is conservatively used
/// as each indivisible protocol-work observed maximum because sqlx does not
/// expose its private socket write/read timings.
pub async fn observe_deployment_db_service(
    config: DatabaseConfig,
    fixed_ca_pem: Vec<u8>,
    identities: DbDeploymentObserverIdentity,
    approved: ApprovedDbReturnBounds,
    samples: u64,
) -> Result<Vec<u8>, DbDeploymentObservationError> {
    if samples == 0
        || fixed_ca_pem.is_empty()
        || [
            identities.environment,
            identities.build,
            identities.gate,
            identities.network_domain,
            identities.credential,
            approved.service_attestation,
            approved.calibration_identity,
        ]
        .contains(&[0; 32])
    {
        return Err(DbDeploymentObservationError::InvalidIdentity);
    }
    validate_observer_url(config.deployment_url())?;
    let base = config
        .options()
        .map_err(|_| DbDeploymentObservationError::InvalidEndpoint)?;
    if base.get_socket().is_some() || base.get_host().is_empty() || base.get_port() == 0 {
        return Err(DbDeploymentObservationError::InvalidEndpoint);
    }
    let host = base.get_host().to_ascii_lowercase();
    let port = base.get_port();
    let database = base
        .get_database()
        .ok_or(DbDeploymentObservationError::InvalidEndpoint)?
        .to_owned();
    let username = base.get_username().to_owned();
    let ca = hash(&fixed_ca_pem);
    let options = base
        .clone()
        .ssl_mode(MySqlSslMode::VerifyIdentity)
        .ssl_ca_from_pem(fixed_ca_pem);
    if !matches!(options.get_ssl_mode(), MySqlSslMode::VerifyIdentity) {
        return Err(DbDeploymentObservationError::InvalidTls);
    }
    let dns = tokio::net::lookup_host((host.as_str(), port))
        .await
        .map_err(|_| DbDeploymentObservationError::ObservationFailed)?
        .map(|address| address.ip())
        .collect::<BTreeSet<IpAddr>>();
    if dns.is_empty() {
        return Err(DbDeploymentObservationError::ObservationFailed);
    }
    let dns_canonical = dns
        .iter()
        .map(IpAddr::to_string)
        .collect::<Vec<_>>()
        .join(",");
    let options_identity = connect_options_identity(&options, ca, identities.credential);
    let mut connection = MySqlConnection::connect_with(&options)
        .await
        .map_err(|_| DbDeploymentObservationError::ObservationFailed)?;
    let tls_cipher = status(&mut connection, "Ssl_cipher").await?;
    let tls_version = status(&mut connection, "Ssl_version").await?;
    if tls_cipher.is_empty() || tls_version.is_empty() {
        return Err(DbDeploymentObservationError::InvalidTls);
    }
    let (version, version_comment, server_hostname): (String, String, String) =
        sqlx::query_as("SELECT @@version, @@version_comment, @@hostname")
            .fetch_one(&mut connection)
            .await
            .map_err(|_| DbDeploymentObservationError::ObservationFailed)?;
    let mut maximum = 0_u64;
    for _ in 0..samples {
        let start = Instant::now();
        connection
            .ping()
            .await
            .map_err(|_| DbDeploymentObservationError::ObservationFailed)?;
        maximum = maximum.max(
            u64::try_from(start.elapsed().as_nanos())
                .map_err(|_| DbDeploymentObservationError::InvalidService)?,
        );
    }
    connection
        .close()
        .await
        .map_err(|_| DbDeploymentObservationError::ObservationFailed)?;
    let server_identity = hash_parts(&[
        version.as_bytes(),
        version_comment.as_bytes(),
        server_hostname.as_bytes(),
        tls_cipher.as_bytes(),
        tls_version.as_bytes(),
    ]);
    let target_system = hash_parts(&[
        std::env::consts::OS.as_bytes(),
        std::env::consts::ARCH.as_bytes(),
    ]);
    let observation = hash_parts(&[
        &options_identity,
        &server_identity,
        &identities.network_domain,
        &approved.service_attestation,
    ]);
    let document = Document {
        schema: SCHEMA.into(),
        release_approval: false,
        endpoint: Endpoint {
            canonical_hostname: host,
            port,
            database_sha256: hash(database.as_bytes()),
            username_sha256: hash(username.as_bytes()),
            credential_identity: identities.credential,
            tls_mode: "verify-identity".into(),
            fixed_ca_sha256: ca,
            connect_options_sha256: options_identity,
            observed_dns_set_sha256: hash(dns_canonical.as_bytes()),
            network_domain_identity: identities.network_domain,
        },
        service: Service {
            tls_cipher_sha256: hash(tls_cipher.as_bytes()),
            tls_version_sha256: hash(tls_version.as_bytes()),
            mariadb_version_sha256: hash(version.as_bytes()),
            version_comment_sha256: hash(version_comment.as_bytes()),
            server_hostname_sha256: hash(server_hostname.as_bytes()),
            server_identity,
        },
        proofs: Proofs {
            normal_return_work_identity: db_normal_return_work_proof().identity(),
            poison_discard_work_identity: db_poison_discard_work_proof().identity(),
            sqlx_identity: hash(SQLX_IDENTITY),
        },
        calibration: Calibration {
            approved_by_gate: true,
            samples,
            max_observed_ping_command_nanos: maximum,
            max_observed_protocol_write_nanos: maximum,
            max_observed_protocol_read_nanos: maximum,
            approved_ping_command_nanos: approved.ping_command_nanos,
            approved_protocol_write_nanos: approved.protocol_write_nanos,
            approved_protocol_read_nanos: approved.protocol_read_nanos,
            calibration_identity: approved.calibration_identity,
            service_attestation: approved.service_attestation,
        },
        identity: Identity {
            environment: identities.environment,
            target_system,
            build: identities.build,
            gate: identities.gate,
            observation,
        },
    };
    validate(&document)?;
    serde_json::to_vec(&document).map_err(|_| DbDeploymentObservationError::InvalidSchema)
}

async fn status(
    connection: &mut MySqlConnection,
    name: &str,
) -> Result<String, DbDeploymentObservationError> {
    let sql = match name {
        "Ssl_cipher" => "SHOW STATUS LIKE 'Ssl_cipher'",
        "Ssl_version" => "SHOW STATUS LIKE 'Ssl_version'",
        _ => return Err(DbDeploymentObservationError::InvalidService),
    };
    let (_, value): (String, String) = sqlx::query_as(sql)
        .fetch_one(connection)
        .await
        .map_err(|_| DbDeploymentObservationError::ObservationFailed)?;
    Ok(value)
}

fn connect_options_identity(
    options: &MySqlConnectOptions,
    ca: [u8; 32],
    credential: [u8; 32],
) -> [u8; 32] {
    hash_parts(&[
        options.get_host().to_ascii_lowercase().as_bytes(),
        &options.get_port().to_be_bytes(),
        options.get_database().unwrap_or("").as_bytes(),
        options.get_username().as_bytes(),
        format!("{:?}", options.get_ssl_mode()).as_bytes(),
        options.get_charset().as_bytes(),
        options.get_collation().unwrap_or("").as_bytes(),
        &ca,
        &credential,
        SQLX_IDENTITY,
        FORCED_OPTIONS_IDENTITY,
    ])
}

fn validate_observer_url(raw: &str) -> Result<(), DbDeploymentObservationError> {
    let Some((scheme, rest)) = raw.split_once("://") else {
        return Err(DbDeploymentObservationError::InvalidEndpoint);
    };
    if scheme != "mysql" || rest.is_empty() || raw.contains('#') {
        return Err(DbDeploymentObservationError::InvalidEndpoint);
    }
    let Some((_, query)) = raw.split_once('?') else {
        return Ok(());
    };
    if query.is_empty() || query.contains('%') || query.contains('?') {
        return Err(DbDeploymentObservationError::InvalidEndpoint);
    }
    let mut seen = BTreeSet::new();
    for pair in query.split('&') {
        let Some((key, value)) = pair.split_once('=') else {
            return Err(DbDeploymentObservationError::InvalidEndpoint);
        };
        if value.is_empty()
            || !seen.insert(key)
            || !matches!(key, "charset" | "collation" | "sslmode" | "ssl-mode")
        {
            // CA material is supplied as bytes and hashed by this observer;
            // all other MySqlConnectOptions overrides are rejected so the
            // canonical identity remains complete and credential-free.
            return Err(DbDeploymentObservationError::InvalidEndpoint);
        }
    }
    Ok(())
}

fn hash(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}
fn hash_parts(parts: &[&[u8]]) -> [u8; 32] {
    let mut h = Sha256::new();
    for part in parts {
        h.update((part.len() as u64).to_be_bytes());
        h.update(part);
    }
    h.finalize().into()
}

impl DeploymentDbServiceAttestation for ObservedDeploymentDbAttestation {
    fn target(&self) -> DbServiceTarget {
        DbServiceTarget::LinuxX86_64
    }
    fn approved(&self) -> bool {
        self.document.calibration.approved_by_gate && !self.document.release_approval
    }
    fn reproducible_measurement(&self) -> bool {
        true
    }
    fn conservative_upper_bound(&self) -> bool {
        true
    }
    fn finite_completion(&self) -> bool {
        self.document.calibration.approved_ping_command_nanos != 0
            && self.document.calibration.approved_protocol_write_nanos != 0
            && self.document.calibration.approved_protocol_read_nanos != 0
    }
    fn work_identity(&self) -> [u8; 32] {
        self.document.proofs.normal_return_work_identity
    }
    fn work_domain(&self) -> DbReturnServiceDomain {
        DbReturnServiceDomain {
            max_ping_commands: 1,
            max_protocol_writes: 1,
            max_protocol_reads: 1,
            max_parallel_returns: 1,
            pool_profile_attested: true,
        }
    }
    fn calibration(&self) -> DbReturnCalibration {
        let c = &self.document.calibration;
        DbReturnCalibration {
            samples: c.samples,
            max_observed_ping_command_nanos: c.max_observed_ping_command_nanos,
            max_observed_protocol_write_nanos: c.max_observed_protocol_write_nanos,
            max_observed_protocol_read_nanos: c.max_observed_protocol_read_nanos,
        }
    }
    fn service_rates(&self) -> DbReturnServiceRates {
        let c = &self.document.calibration;
        DbReturnServiceRates {
            ping_command_nanos: c.approved_ping_command_nanos,
            protocol_write_nanos: c.approved_protocol_write_nanos,
            protocol_read_nanos: c.approved_protocol_read_nanos,
        }
    }
    fn calibration_identity(&self) -> [u8; 32] {
        self.document.calibration.calibration_identity
    }
    fn environment_identity(&self) -> [u8; 32] {
        self.document.identity.environment
    }
    fn database_server_identity(&self) -> [u8; 32] {
        self.document.service.server_identity
    }
    fn network_identity(&self) -> [u8; 32] {
        self.document.endpoint.network_domain_identity
    }
    fn sqlx_identity(&self) -> [u8; 32] {
        self.document.proofs.sqlx_identity
    }
    fn target_system_identity(&self) -> [u8; 32] {
        self.document.identity.target_system
    }
    fn service_attestation(&self) -> [u8; 32] {
        self.document.calibration.service_attestation
    }
    fn build_identity(&self) -> [u8; 32] {
        self.document.identity.build
    }
    fn gate_identity(&self) -> [u8; 32] {
        self.document.identity.gate
    }
}

#[cfg(test)]
mod tests {
    use std::{env, fs};

    use crate::c1_service_attestation::{
        DbDeploymentIdentity, derive_db_return_service, verify_db_service,
    };

    use super::*;

    fn identity() -> DbDeploymentObserverIdentity {
        DbDeploymentObserverIdentity {
            environment: [1; 32],
            build: [2; 32],
            gate: [3; 32],
            network_domain: [4; 32],
            credential: [5; 32],
        }
    }

    fn bounds() -> ApprovedDbReturnBounds {
        ApprovedDbReturnBounds {
            // Evidence-only ceiling, deliberately external to observation.
            ping_command_nanos: 10_000_000_000,
            protocol_write_nanos: 10_000_000_000,
            protocol_read_nanos: 10_000_000_000,
            service_attestation: [6; 32],
            calibration_identity: [7; 32],
        }
    }

    #[tokio::test]
    async fn real_tls_mariadb_observation_parses_and_derives_checked_proof() {
        let (Ok(url), Ok(ca_path)) = (
            env::var("SADDLE_TEST_TLS_DATABASE_URL"),
            env::var("SADDLE_TEST_TLS_DATABASE_CA"),
        ) else {
            eprintln!("skipping TLS DB observation: environment is not set");
            return;
        };
        let bytes = observe_deployment_db_service(
            DatabaseConfig::new(url),
            fs::read(ca_path).unwrap(),
            identity(),
            bounds(),
            8,
        )
        .await
        .unwrap();
        let attestation = parse_deployment_db_observation(&bytes).unwrap();
        let document: Document = serde_json::from_slice(&bytes).unwrap();
        assert!(!document.release_approval);
        assert_eq!(document.endpoint.tls_mode, "verify-identity");
        assert!(!String::from_utf8_lossy(&bytes).contains("root@"));
        let expected = DbDeploymentIdentity {
            target: DbServiceTarget::LinuxX86_64,
            environment: identity().environment,
            database_server: document.service.server_identity,
            network: identity().network_domain,
            sqlx: document.proofs.sqlx_identity,
            target_system: document.identity.target_system,
            build: identity().build,
            gate: identity().gate,
        };
        let verified =
            verify_db_service(attestation, expected, db_normal_return_work_proof()).unwrap();
        assert_eq!(
            derive_db_return_service(verified)
                .unwrap()
                .max_service_nanos(),
            30_000_000_000
        );
    }

    #[test]
    fn strict_schema_missing_bound_unknown_field_and_unsafe_tls_reject() {
        let document = Document {
            schema: SCHEMA.into(),
            release_approval: false,
            endpoint: Endpoint {
                canonical_hostname: "db.example.test".into(),
                port: 3306,
                database_sha256: [1; 32],
                username_sha256: [2; 32],
                credential_identity: [3; 32],
                tls_mode: "verify-identity".into(),
                fixed_ca_sha256: [4; 32],
                connect_options_sha256: [5; 32],
                observed_dns_set_sha256: [6; 32],
                network_domain_identity: [7; 32],
            },
            service: Service {
                tls_cipher_sha256: [8; 32],
                tls_version_sha256: [9; 32],
                mariadb_version_sha256: [10; 32],
                version_comment_sha256: [11; 32],
                server_hostname_sha256: [12; 32],
                server_identity: [13; 32],
            },
            proofs: Proofs {
                normal_return_work_identity: db_normal_return_work_proof().identity(),
                poison_discard_work_identity: db_poison_discard_work_proof().identity(),
                sqlx_identity: hash(SQLX_IDENTITY),
            },
            calibration: Calibration {
                approved_by_gate: true,
                samples: 1,
                max_observed_ping_command_nanos: 10,
                max_observed_protocol_write_nanos: 10,
                max_observed_protocol_read_nanos: 10,
                approved_ping_command_nanos: 20,
                approved_protocol_write_nanos: 20,
                approved_protocol_read_nanos: 20,
                calibration_identity: [14; 32],
                service_attestation: [15; 32],
            },
            identity: Identity {
                environment: [16; 32],
                target_system: [17; 32],
                build: [18; 32],
                gate: [19; 32],
                observation: [20; 32],
            },
        };
        let mut value = serde_json::to_value(&document).unwrap();
        value["unknown"] = serde_json::json!(true);
        assert_eq!(
            parse_deployment_db_observation(&serde_json::to_vec(&value).unwrap()).unwrap_err(),
            DbDeploymentObservationError::InvalidSchema
        );
        let mut value = serde_json::to_value(&document).unwrap();
        value["calibration"]["approved_ping_command_nanos"] = serde_json::json!(0);
        assert_eq!(
            parse_deployment_db_observation(&serde_json::to_vec(&value).unwrap()).unwrap_err(),
            DbDeploymentObservationError::MissingApprovedBound
        );
        let mut value = serde_json::to_value(&document).unwrap();
        value["endpoint"]["tls_mode"] = serde_json::json!("required");
        assert_eq!(
            parse_deployment_db_observation(&serde_json::to_vec(&value).unwrap()).unwrap_err(),
            DbDeploymentObservationError::InvalidEndpoint
        );
        let mut value = serde_json::to_value(&document).unwrap();
        value["proofs"]["poison_discard_work_identity"] = serde_json::json!(vec![99; 32]);
        assert_eq!(
            parse_deployment_db_observation(&serde_json::to_vec(&value).unwrap()).unwrap_err(),
            DbDeploymentObservationError::InvalidIdentity
        );
        assert!(validate_observer_url("mysql://u:p@db.example/d?charset=utf8mb4").is_ok());
        assert_eq!(
            validate_observer_url("mysql://u:p@db.example/d?ssl-cert=client.pem").unwrap_err(),
            DbDeploymentObservationError::InvalidEndpoint
        );
    }
}