Plan: canopy-store upload validation — full port (Issue #435)

// crates/canopy-store/src/scanner.rs

use async_trait::async_trait;

#[async_trait]
pub trait Scanner: Send + Sync {
    async fn scan(&self, bytes: &[u8]) -> Result<ScanResult, ScanError>;
    fn name(&self) -> &'static str;
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ScanResult {
    Clean,
    Infected { signature: String },
    Skipped { reason: String },
}

#[derive(Debug, thiserror::Error)]
pub enum ScanError {
    #[error("scanner backend error: {0}")]
    Backend(String),
    #[error("scan timed out after {0:?}")]
    Timeout(std::time::Duration),
}

pub struct NoopScanner;

#[async_trait]
impl Scanner for NoopScanner {
    async fn scan(&self, _bytes: &[u8]) -> Result<ScanResult, ScanError> {
        Ok(ScanResult::Clean)
    }
    fn name(&self) -> &'static str { "noop" }
}

// Additions to crates/canopy-store/src/error.rs

/// Upload contained zero bytes.
Empty,

/// Magic-byte sniffing could not identify the content type.
UnknownContentType,

/// Caller-claimed content type does not match the magic-byte-detected
/// type. Catches client-lies and internal rendering bugs.
ContentTypeMismatch { claimed: String, actual: String },

/// AV scanner reported a positive detection.
Infected { signature: String, scanner: String },

/// Scanner backend failed (transport-level, not a positive detection).
ScannerError(#[from] crate::scanner::ScanError),

pub struct ValidatedUpload {
    pub sha256: [u8; 32],
    pub content_type: String,
    pub size: usize,
    pub sanitized_filename: Option<String>,
}

pub async fn validate_upload(
    bytes: &[u8],
    claimed_content_type: &str,
    filename: Option<&str>,
    validation: &UploadValidation<'_>,
    scanner: &dyn crate::scanner::Scanner,
) -> Result<ValidatedUpload, StoreError> {
    if bytes.is_empty() { return Err(StoreError::Empty); }
    if bytes.len() > validation.max_bytes {
        return Err(StoreError::TooLarge { size: bytes.len(), limit: validation.max_bytes });
    }
    let kind = infer::get(bytes).ok_or(StoreError::UnknownContentType)?;
    let actual = kind.mime_type();
    if actual != claimed_content_type {
        return Err(StoreError::ContentTypeMismatch {
            claimed: claimed_content_type.to_string(),
            actual: actual.to_string(),
        });
    }
    if !validation.allowed_mime_types.contains(&actual) {
        return Err(StoreError::DisallowedContentType(actual.to_string()));
    }
    use sha2::{Sha256, Digest};
    let sha256: [u8; 32] = Sha256::digest(bytes).into();
    use crate::scanner::ScanResult;
    match scanner.scan(bytes).await? {
        ScanResult::Clean | ScanResult::Skipped { .. } => {}
        ScanResult::Infected { signature } => {
            return Err(StoreError::Infected {
                signature,
                scanner: scanner.name().to_string(),
            });
        }
    }
    let sanitized_filename = filename.map(sanitize_filename).transpose()?;
    Ok(ValidatedUpload { sha256, content_type: actual.to_string(), size: bytes.len(), sanitized_filename })
}

pub fn sanitize_filename(name: &str) -> Result<String, StoreError> {
    use unicode_normalization::UnicodeNormalization;
    let normalized: String = name.nfc().collect();
    for c in normalized.chars() {
        let cp = c as u32;
        let is_c0 = cp <= 0x1F;
        let is_c1 = (0x7F..=0x9F).contains(&cp);
        let is_bidi = matches!(cp, 0x202A..=0x202E | 0x2066..=0x2069);
        if is_c0 || is_c1 || is_bidi {
            return Err(StoreError::InvalidFilename(
                format!("disallowed character U+{cp:04X}")
            ));
        }
    }
    let stripped: String = normalized.chars()
        .filter(|c| !matches!(*c, '/' | '\\'))
        .collect();
    let trimmed = stripped.trim().trim_matches('.').to_string();
    if trimmed.is_empty() {
        return Err(StoreError::InvalidFilename("empty after sanitization".into()));
    }
    let mut out = trimmed;
    if out.len() > 255 {
        let mut cut = 255;
        while !out.is_char_boundary(cut) { cut -= 1; }
        out.truncate(cut);
    }
    Ok(out)
}

// crates/canopy-store/src/disposition.rs

pub fn content_disposition_header(filename: &str, disposition: Disposition) -> String {
    let kind = match disposition {
        Disposition::Inline => "inline",
        Disposition::Attachment => "attachment",
    };
    let is_ascii_safe = filename.is_ascii()
        && !filename.chars().any(|c| c == '"' || c == '\\');
    if is_ascii_safe {
        format!(r#"{kind}; filename="{filename}""#)
    } else {
        format!("{kind}; filename*=UTF-8''{}", percent_encode(filename))
    }
}

pub enum Disposition { Inline, Attachment }

fn percent_encode(s: &str) -> String {
    let mut out = String::with_capacity(s.len() * 3);
    for b in s.bytes() {
        let unreserved = b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~');
        if unreserved { out.push(b as char); }
        else { use std::fmt::Write; write!(out, "%{b:02X}").unwrap(); }
    }
    out
}

// crates/canopy-store/src/reconcile.rs

use std::collections::HashSet;

pub struct ReconcileReport {
    pub orphans: Vec<String>,  // in DB, not in storage
    pub leaks: Vec<String>,    // in storage, not in DB
}

pub fn compare(db_paths: &[String], live_keys: &[String]) -> ReconcileReport {
    let db: HashSet<&str> = db_paths.iter().map(String::as_str).collect();
    let live: HashSet<&str> = live_keys.iter().map(String::as_str).collect();
    ReconcileReport {
        orphans: db.difference(&live).map(|s| s.to_string()).collect(),
        leaks: live.difference(&db).map(|s| s.to_string()).collect(),
    }
}

impl Store {
    #[deprecated(note = "use put_validated; #435 requires upload validation for all writes")]
    pub async fn put(&self, path: &str, data: Bytes) -> Result<(), StoreError> { /* unchanged */ }

    pub async fn put_validated(
        &self,
        path: &str,
        data: Bytes,
        claimed_content_type: &str,
        validation: &UploadValidation<'_>,
        scanner: &dyn Scanner,
    ) -> Result<ValidatedUpload, StoreError> {
        let validated = validate_upload(&data, claimed_content_type, None, validation, scanner).await?;
        #[allow(deprecated)]
        self.put(path, data).await?;
        Ok(validated)
    }

    pub async fn list_all_keys(&self, prefix: Option<&str>) -> Result<Vec<String>, StoreError> {
        // Wraps object_store::list with full pagination.
    }
}

let validation = UploadValidation::default();
let validated = self.object_store
    .put_validated(&path, Bytes::from(rendered.pdf_bytes), "application/pdf",
                   &validation, &*self.scanner)
    .await?;
// validated.sha256 + content_type flow into the notices row via
// insert_notice (which gains 2 new column args).

// services/canopy-notices/src/reconcile.rs

pub async fn run_reconciliation_loop(store: Arc<Store>, db: PgPool, interval: Duration) {
    let mut ticker = tokio::time::interval(interval);
    ticker.tick().await; // skip immediate first tick
    loop {
        ticker.tick().await;
        if let Err(e) = reconcile_once(&store, &db).await {
            tracing::warn!(error = %e, "notice reconciliation failed");
        }
    }
}

-- services/canopy-notices/migrations/20260515000000_add_notice_content_integrity.sql

ALTER TABLE notices
    ADD COLUMN content_sha256 BYTEA NOT NULL DEFAULT '\x',
    ADD COLUMN content_type TEXT NOT NULL DEFAULT 'application/pdf',
    ADD COLUMN scan_status TEXT NOT NULL DEFAULT 'noop';

COMMENT ON COLUMN notices.content_sha256 IS 'SHA-256 of the stored PDF bytes (#435 / Pub 1075 §9 integrity chain).';
COMMENT ON COLUMN notices.content_type IS 'Magic-byte-verified content type at upload time (#435).';
COMMENT ON COLUMN notices.scan_status IS 'AV scanner result at upload: clean | infected | skipped | noop (#435).';