feat(M8+M9): CA certificates page + Reporting CSV/PDF with charts

crates/pm-ca/Cargo.toml

@@ -6,11 +6,20 @@ authors.workspace = true
 license.workspace = true
 
 [dependencies]
-pm-core = { path = "../pm-core" }
-tokio = { workspace = true }
-serde = { workspace = true }
+pm-core    = { path = "../pm-core" }
+tokio      = { workspace = true }
+serde      = { workspace = true }
 serde_json = { workspace = true }
-thiserror = { workspace = true }
-anyhow = { workspace = true }
-tracing = { workspace = true }
-chrono = { workspace = true }
+thiserror  = { workspace = true }
+anyhow     = { workspace = true }
+tracing    = { workspace = true }
+chrono     = { workspace = true }
+sqlx       = { workspace = true }
+uuid       = { workspace = true }
+rand       = { workspace = true }
+hex        = { workspace = true }
+sha2       = { workspace = true }
+rustls     = { workspace = true }
+rcgen      = { workspace = true }
+pem        = { workspace = true }
+time       = { workspace = true }

crates/pm-ca/src/ca.rs

@@ -1 +1,420 @@
-//! Internal CA stub for M8.
+//! Internal Certificate Authority for Linux Patch Manager.
+//!
+//! Issues and renews mTLS client certificates for agent communication.
+//! Uses rcgen (ECDSA P-256) for all certificate generation.
+//! CA key and certificate are stored on disk under `base_dir`
+//! (default: /etc/patch-manager/ca/).
+//! Certificate metadata is persisted in the `certificates` PostgreSQL table.
+
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, Duration as ChronoDuration, Utc};
+use rand::RngCore;
+use rcgen::{
+    BasicConstraints, Certificate, CertificateParams, DistinguishedName, DnType,
+    ExtendedKeyUsagePurpose, Ia5String, IsCa, KeyPair, KeyUsagePurpose, SanType,
+    SerialNumber, PKCS_ECDSA_P256_SHA256,
+};
+use sqlx::{PgPool, Row};
+use time::{Duration as TimeDuration, OffsetDateTime};
+use uuid::Uuid;
+
+// ---------------------------------------------------------------------------
+// Public types
+// ---------------------------------------------------------------------------
+
+/// Returned by [`CertAuthority::issue_client_cert`] and [`CertAuthority::renew_cert`].
+///
+/// The private key is intentionally **not** stored in the database.
+#[derive(Debug, Clone)]
+pub struct IssuedCert {
+    /// PEM-encoded public certificate.
+    pub cert_pem: String,
+    /// PEM-encoded private key (PKCS#8).
+    pub key_pem: String,
+    /// Hex-encoded 16-byte random serial number.
+    pub serial_number: String,
+    /// Certificate expiry timestamp (UTC).
+    pub expires_at: DateTime<Utc>,
+}
+
+// ---------------------------------------------------------------------------
+// CertAuthority
+// ---------------------------------------------------------------------------
+
+/// Thread-safe, cloneable handle to the internal certificate authority.
+///
+/// CA certificate and key are held in memory as PEM strings; rcgen objects
+/// are reconstructed on demand so this struct is unconditionally `Send + Sync`.
+#[derive(Debug, Clone)]
+pub struct CertAuthority {
+    #[allow(dead_code)]
+    base_dir: PathBuf,
+    /// PEM-encoded CA certificate (public cert only).
+    ca_cert_pem: String,
+    /// PEM-encoded CA private key (PKCS#8).
+    ca_key_pem: String,
+}
+
+// ---------------------------------------------------------------------------
+// Private helpers
+// ---------------------------------------------------------------------------
+
+/// Generate a 16-byte cryptographically-random serial number.
+/// Returns `(rcgen::SerialNumber, hex_encoded_string)`.
+fn make_serial() -> (SerialNumber, String) {
+    let mut bytes = [0u8; 16];
+    rand::rngs::OsRng.fill_bytes(&mut bytes);
+    let hex_serial = hex::encode(bytes);
+    let serial = SerialNumber::from_slice(&bytes);
+    (serial, hex_serial)
+}
+
+/// `OffsetDateTime::now_utc()` offset forward by `days` (for rcgen params).
+fn odt_offset_days(days: i64) -> OffsetDateTime {
+    OffsetDateTime::now_utc() + TimeDuration::days(days)
+}
+
+/// `chrono::Utc::now()` offset forward by `days` (for DB / return values).
+fn chrono_offset_days(days: i64) -> DateTime<Utc> {
+    Utc::now() + ChronoDuration::days(days)
+}
+
+/// Build a `CertificateParams` with common fields pre-filled.
+/// Caller still needs to set `is_ca`, `key_usages`, `extended_key_usages`, and `subject_alt_names`.
+fn base_params(
+    cn: &str,
+    validity_days: i64,
+) -> Result<(CertificateParams, String, DateTime<Utc>)> {
+    let (serial, serial_hex) = make_serial();
+    let expires_at = chrono_offset_days(validity_days);
+
+    let mut params = CertificateParams::default();
+    params.not_before = OffsetDateTime::now_utc();
+    params.not_after = odt_offset_days(validity_days);
+    params.serial_number = Some(serial);
+
+    let mut dn = DistinguishedName::new();
+    dn.push(DnType::CommonName, cn);
+    params.distinguished_name = dn;
+
+    Ok((params, serial_hex, expires_at))
+}
+
+/// Write `contents` to `path` and set Unix permissions to `0600`.
+async fn write_protected(path: &Path, contents: &str) -> Result<()> {
+    use std::os::unix::fs::PermissionsExt;
+    tokio::fs::write(path, contents).await?;
+    let perms = std::fs::Permissions::from_mode(0o600);
+    tokio::fs::set_permissions(path, perms).await?;
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// impl CertAuthority
+// ---------------------------------------------------------------------------
+
+impl CertAuthority {
+    // -----------------------------------------------------------------------
+    // Construction
+    // -----------------------------------------------------------------------
+
+    /// Load an existing CA from disk, or generate a brand-new one if absent.
+    ///
+    /// Files managed:
+    /// * `{base_dir}/ca.key`  — PKCS#8 PEM private key (mode `0600`)
+    /// * `{base_dir}/ca.crt`  — PEM certificate         (mode `0600`)
+    ///
+    /// On first generation the CA row is inserted into `certificates`
+    /// with `host_id = NULL` (marks it as the root CA record).
+    pub async fn init(base_dir: &Path, db: &PgPool) -> Result<Self> {
+        let key_path = base_dir.join("ca.key");
+        let crt_path = base_dir.join("ca.crt");
+
+        // ── Load existing CA ──────────────────────────────────────────────
+        if key_path.exists() && crt_path.exists() {
+            tracing::info!(path = %base_dir.display(), "Loading existing root CA from disk");
+
+            let ca_key_pem = tokio::fs::read_to_string(&key_path)
+                .await
+                .context("read ca.key")?;
+            let ca_cert_pem = tokio::fs::read_to_string(&crt_path)
+                .await
+                .context("read ca.crt")?;
+
+            // Validate that both PEMs parse without error.
+            KeyPair::from_pem(&ca_key_pem)
+                .context("parse CA private-key PEM")?;
+            CertificateParams::from_ca_cert_pem(&ca_cert_pem)
+                .context("parse CA certificate PEM")?;
+
+            tracing::info!("Root CA loaded successfully");
+            return Ok(Self {
+                base_dir: base_dir.to_owned(),
+                ca_cert_pem,
+                ca_key_pem,
+            });
+        }
+
+        // ── Generate new CA ───────────────────────────────────────────────
+        tracing::info!(
+            path = %base_dir.display(),
+            "Generating new root CA (ECDSA P-256, 10-year validity)"
+        );
+        tokio::fs::create_dir_all(base_dir)
+            .await
+            .context("create CA directory")?;
+
+        let ca_key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256)
+            .context("generate CA key pair")?;
+
+        let (serial, serial_hex) = make_serial();
+        let expires_at = chrono_offset_days(365 * 10);
+
+        let mut params = CertificateParams::default();
+        params.not_before = OffsetDateTime::now_utc();
+        params.not_after = odt_offset_days(365 * 10);
+        params.serial_number = Some(serial);
+        params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
+        params.key_usages = vec![
+            KeyUsagePurpose::KeyCertSign,
+            KeyUsagePurpose::CrlSign,
+        ];
+
+        let mut dn = DistinguishedName::new();
+        dn.push(DnType::CommonName, "Patch Manager Root CA");
+        dn.push(DnType::OrganizationName, "Patch Manager");
+        params.distinguished_name = dn;
+
+        let ca_cert_obj = params.self_signed(&ca_key)
+            .context("self-sign CA certificate")?;
+        let ca_cert_pem = ca_cert_obj.pem();
+        let ca_key_pem  = ca_key.serialize_pem();
+
+        write_protected(&key_path, &ca_key_pem)
+            .await
+            .context("write ca.key")?;
+        write_protected(&crt_path, &ca_cert_pem)
+            .await
+            .context("write ca.crt")?;
+
+        tracing::info!(
+            serial = %serial_hex,
+            expires_at = %expires_at,
+            "Root CA generated and written to disk"
+        );
+
+        // Persist CA cert metadata (host_id = NULL marks the root CA row).
+        sqlx::query(
+            "INSERT INTO certificates \
+             (host_id, serial_number, common_name, status, expires_at, cert_pem) \
+             VALUES (NULL, $1, 'Patch Manager Root CA', 'active'::cert_status, $2, $3)",
+        )
+        .bind(&serial_hex)
+        .bind(expires_at)
+        .bind(&ca_cert_pem)
+        .execute(db)
+        .await
+        .context("insert root CA cert into database")?;
+
+        tracing::info!("Root CA certificate recorded in database");
+
+        Ok(Self {
+            base_dir: base_dir.to_owned(),
+            ca_cert_pem,
+            ca_key_pem,
+        })
+    }
+
+    // -----------------------------------------------------------------------
+    // Public accessors
+    // -----------------------------------------------------------------------
+
+    /// Return the PEM-encoded root CA certificate (public cert only).
+    pub fn root_cert_pem(&self) -> &str {
+        &self.ca_cert_pem
+    }
+
+    // -----------------------------------------------------------------------
+    // Certificate issuance
+    // -----------------------------------------------------------------------
+
+    /// Issue a one-year mTLS client certificate for a managed host.
+    ///
+    /// * Subject: `CN=<hostname>`
+    /// * Key usage: Digital Signature
+    /// * Extended key usage: Client Authentication
+    ///
+    /// The certificate PEM is stored in `certificates`.
+    /// The private key is returned to the caller **only** and never persisted.
+    pub async fn issue_client_cert(
+        &self,
+        host_id: Uuid,
+        hostname: &str,
+        db: &PgPool,
+    ) -> Result<IssuedCert> {
+        tracing::info!(host_id = %host_id, hostname, "Issuing mTLS client certificate");
+
+        let key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256)
+            .context("generate client key pair")?;
+
+        let (mut params, serial_hex, expires_at) = base_params(hostname, 365)?;
+        params.is_ca = IsCa::ExplicitNoCa;
+        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
+        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
+
+        let (ca_key, ca_cert) = self.ca_objects()?;
+        let cert = params
+            .signed_by(&key, &ca_cert, &ca_key)
+            .context("sign client cert with CA")?;
+
+        let cert_pem = cert.pem();
+        let key_pem  = key.serialize_pem();
+
+        sqlx::query(
+            "INSERT INTO certificates \
+             (host_id, serial_number, common_name, status, expires_at, cert_pem) \
+             VALUES ($1, $2, $3, 'active'::cert_status, $4, $5)",
+        )
+        .bind(host_id)
+        .bind(&serial_hex)
+        .bind(hostname)
+        .bind(expires_at)
+        .bind(&cert_pem)
+        .execute(db)
+        .await
+        .context("insert client cert into database")?;
+
+        tracing::info!(
+            host_id = %host_id,
+            hostname,
+            serial = %serial_hex,
+            expires_at = %expires_at,
+            "Client certificate issued successfully"
+        );
+
+        Ok(IssuedCert { cert_pem, key_pem, serial_number: serial_hex, expires_at })
+    }
+
+    /// Revoke a certificate by database ID.
+    ///
+    /// Sets `status = 'revoked'` and `revoked_at = NOW()` in the `certificates` table.
+    /// Does **not** reissue a replacement; use [`renew_cert`] for that.
+    pub async fn revoke_cert(&self, cert_id: Uuid, db: &PgPool) -> Result<()> {
+        tracing::info!(cert_id = %cert_id, "Revoking certificate");
+
+        let rows = sqlx::query(
+            "UPDATE certificates \
+             SET status = 'revoked'::cert_status, revoked_at = NOW() \
+             WHERE id = $1",
+        )
+        .bind(cert_id)
+        .execute(db)
+        .await
+        .context("revoke certificate in database")?;
+
+        if rows.rows_affected() == 0 {
+            anyhow::bail!("certificate not found: {}", cert_id);
+        }
+
+        tracing::info!(cert_id = %cert_id, "Certificate revoked");
+        Ok(())
+    }
+
+    /// Renew a certificate: revoke the existing cert and issue a new one with
+    /// the same `host_id` and `common_name`.
+    pub async fn renew_cert(&self, cert_id: Uuid, db: &PgPool) -> Result<IssuedCert> {
+        tracing::info!(cert_id = %cert_id, "Renewing certificate");
+
+        // Fetch the existing cert's host_id and common_name.
+        let row = sqlx::query(
+            "SELECT host_id, common_name FROM certificates WHERE id = $1",
+        )
+        .bind(cert_id)
+        .fetch_one(db)
+        .await
+        .context("fetch certificate for renewal")?;
+
+        let host_id: Uuid = row.try_get("host_id")
+            .context("certificate has no host_id (cannot renew root CA)")?;
+        let common_name: String = row.try_get("common_name")
+            .context("fetch common_name")?;
+
+        // Revoke the old cert first.
+        self.revoke_cert(cert_id, db).await?;
+
+        // Issue a fresh cert with the same CN.
+        let issued = self.issue_client_cert(host_id, &common_name, db).await?;
+
+        tracing::info!(
+            old_cert_id = %cert_id,
+            new_serial   = %issued.serial_number,
+            "Certificate renewed"
+        );
+        Ok(issued)
+    }
+
+    /// Generate a self-signed TLS certificate for the web UI using the CA.
+    ///
+    /// * Subject: `CN=<hostname>`
+    /// * Key usage: Digital Signature
+    /// * Extended key usage: Server Authentication
+    /// * SAN: DNS `<hostname>`
+    ///
+    /// Returns `(cert_pem, key_pem)`. This certificate is **not** stored in the
+    /// database; it is intended for runtime use only.
+    pub async fn issue_web_tls_cert(
+        &self,
+        hostname: &str,
+    ) -> Result<(String, String)> {
+        tracing::info!(hostname, "Issuing web TLS certificate");
+
+        let key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256)
+            .context("generate web TLS key pair")?;
+
+        let (mut params, serial_hex, expires_at) = base_params(hostname, 365)?;
+        params.is_ca = IsCa::ExplicitNoCa;
+        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
+        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
+        params.subject_alt_names = vec![SanType::DnsName(
+            Ia5String::try_from(hostname.to_owned()).context("hostname is not valid IA5")?,
+        )];
+
+        let (ca_key, ca_cert) = self.ca_objects()?;
+        let cert = params
+            .signed_by(&key, &ca_cert, &ca_key)
+            .context("sign web TLS cert with CA")?;
+
+        let cert_pem = cert.pem();
+        let key_pem  = key.serialize_pem();
+
+        tracing::info!(
+            hostname,
+            serial = %serial_hex,
+            expires_at = %expires_at,
+            "Web TLS certificate issued"
+        );
+
+        Ok((cert_pem, key_pem))
+    }
+
+    // -----------------------------------------------------------------------
+    // Private helpers
+    // -----------------------------------------------------------------------
+
+    /// Reconstruct rcgen `(KeyPair, Certificate)` from the in-memory PEM strings.
+    ///
+    /// The returned `Certificate` is used solely as an issuer reference when
+    /// signing leaf certificates; it is never distributed directly.
+    fn ca_objects(&self) -> Result<(KeyPair, Certificate)> {
+        let key = KeyPair::from_pem(&self.ca_key_pem)
+            .context("reconstruct CA key pair from PEM")?;
+        let params = CertificateParams::from_ca_cert_pem(&self.ca_cert_pem)
+            .context("reconstruct CA params from PEM")?;
+        let cert = params
+            .self_signed(&key)
+            .context("reconstruct CA certificate for signing")?;
+        Ok((key, cert))
+    }
+}

crates/pm-ca/src/lib.rs

@@ -1,7 +1,7 @@
 //! pm-ca — Internal Certificate Authority.
 //!
 //! Issues and renews mTLS client certificates for agent communication.
-//! Uses rcgen + rustls. CA key stored at /etc/patch-manager/ca/ca.key.
-//!
-//! M1: Stub. Full implementation in M8.
+//! Uses rcgen + rustls. CA key stored at /etc/patch-manager/ca/.
+
 pub mod ca;
+pub use ca::{CertAuthority, IssuedCert};

crates/pm-reports/Cargo.toml

@@ -14,3 +14,12 @@ thiserror = { workspace = true }
 anyhow = { workspace = true }
 tracing = { workspace = true }
 chrono = { workspace = true }
+sqlx = { workspace = true }
+uuid = { workspace = true }
+
+# Report generation
+csv = "1"
+printpdf = { version = "0.7", features = ["embedded_images"] }
+plotters = { version = "0.3", default-features = false, features = ["bitmap_backend", "bitmap_encoder", "line_series", "area_series"] }
+plotters-bitmap = { version = "0.3" }
+image = { version = "0.25", default-features = false, features = ["png"] }

crates/pm-reports/src/csv.rs

@@ -1 +1,334 @@
-//! csv report generation stub for M9.
+//! CSV report generation for pm-reports.
+
+use crate::{ReportParams, ReportType};
+use anyhow::Context;
+
+/// Generate a CSV report and return the raw bytes.
+pub async fn generate_csv(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    match params.report_type {
+        ReportType::Compliance => compliance_csv(pool, params).await,
+        ReportType::PatchHistory => patch_history_csv(pool, params).await,
+        ReportType::Vulnerability => vulnerability_csv(pool, params).await,
+        ReportType::Audit => audit_csv(pool, params).await,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Compliance
+// ---------------------------------------------------------------------------
+
+async fn compliance_csv(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    let rows = if let Some(gid) = params.group_id {
+        sqlx::query("
+SELECT
+    h.id::text            AS host_id,
+    h.display_name,
+    h.fqdn,
+    h.health_status::text AS health_status,
+    h.last_patch_at,
+    COALESCE(pd.total_packages, 0)  AS total_packages,
+    COALESCE(pd.pending_patches, 0) AS pending_patches,
+    CASE WHEN COALESCE(pd.total_packages,0) = 0 THEN 100.0
+         ELSE ROUND((1.0 - pd.pending_patches::float / NULLIF(pd.total_packages,0)) * 100, 1)
+    END AS compliance_pct,
+    COALESCE(string_agg(DISTINCT g.name, ', '), '') AS group_names
+FROM hosts h
+LEFT JOIN host_patch_data pd ON pd.host_id = h.id
+LEFT JOIN host_groups hg ON hg.host_id = h.id
+LEFT JOIN groups g ON g.id = hg.group_id
+WHERE h.id IN (
+    SELECT host_id FROM host_groups WHERE group_id = $1
+)
+GROUP BY h.id, pd.total_packages, pd.pending_patches
+ORDER BY compliance_pct ASC
+")
+        .bind(gid)
+        .fetch_all(pool)
+        .await
+        .context("compliance query (group filter) failed")?
+    } else {
+        sqlx::query("
+SELECT
+    h.id::text            AS host_id,
+    h.display_name,
+    h.fqdn,
+    h.health_status::text AS health_status,
+    h.last_patch_at,
+    COALESCE(pd.total_packages, 0)  AS total_packages,
+    COALESCE(pd.pending_patches, 0) AS pending_patches,
+    CASE WHEN COALESCE(pd.total_packages,0) = 0 THEN 100.0
+         ELSE ROUND((1.0 - pd.pending_patches::float / NULLIF(pd.total_packages,0)) * 100, 1)
+    END AS compliance_pct,
+    COALESCE(string_agg(DISTINCT g.name, ', '), '') AS group_names
+FROM hosts h
+LEFT JOIN host_patch_data pd ON pd.host_id = h.id
+LEFT JOIN host_groups hg ON hg.host_id = h.id
+LEFT JOIN groups g ON g.id = hg.group_id
+GROUP BY h.id, pd.total_packages, pd.pending_patches
+ORDER BY compliance_pct ASC
+")
+        .fetch_all(pool)
+        .await
+        .context("compliance query failed")?
+    };
+
+    let mut wtr = csv::Writer::from_writer(vec![]);
+    wtr.write_record(&[
+        "host_id", "display_name", "fqdn", "group_names",
+        "total_packages", "pending_patches", "compliance_pct",
+        "last_patch_at", "health_status",
+    ])?;
+
+    for row in &rows {
+        use sqlx::Row;
+        let host_id: String          = row.try_get("host_id").unwrap_or_default();
+        let display_name: String     = row.try_get("display_name").unwrap_or_default();
+        let fqdn: String             = row.try_get("fqdn").unwrap_or_default();
+        let group_names: String      = row.try_get("group_names").unwrap_or_default();
+        let total_packages: i64      = row.try_get("total_packages").unwrap_or(0);
+        let pending_patches: i64     = row.try_get("pending_patches").unwrap_or(0);
+        let compliance_pct: f64      = row.try_get("compliance_pct").unwrap_or(0.0);
+        let last_patch_at: Option<chrono::DateTime<chrono::Utc>> =
+            row.try_get("last_patch_at").unwrap_or(None);
+        let health_status: String    = row.try_get("health_status").unwrap_or_default();
+
+        wtr.write_record(&[
+            host_id,
+            display_name,
+            fqdn,
+            group_names,
+            total_packages.to_string(),
+            pending_patches.to_string(),
+            format!("{:.1}", compliance_pct),
+            last_patch_at.map(|d| d.to_rfc3339()).unwrap_or_default(),
+            health_status,
+        ])?;
+    }
+
+    Ok(wtr.into_inner().context("csv flush failed")?)
+}
+
+// ---------------------------------------------------------------------------
+// Patch history
+// ---------------------------------------------------------------------------
+
+async fn patch_history_csv(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    let rows = sqlx::query("
+SELECT
+    pj.id::text        AS job_id,
+    pj.kind::text      AS job_kind,
+    pj.status::text    AS job_status,
+    h.display_name,
+    h.fqdn,
+    jsonb_array_length(COALESCE(pj.patch_selection->'packages', '[]'::jsonb)) AS package_count,
+    pjh.started_at,
+    pjh.completed_at,
+    EXTRACT(EPOCH FROM (pjh.completed_at - pjh.started_at))::bigint AS duration_seconds,
+    COALESCE(u.username, 'system') AS operator
+FROM patch_job_hosts pjh
+JOIN patch_jobs pj ON pj.id = pjh.job_id
+JOIN hosts h ON h.id = pjh.host_id
+LEFT JOIN users u ON u.id = pj.created_by_user_id
+WHERE ($1::timestamptz IS NULL OR pjh.started_at >= $1)
+  AND ($2::timestamptz IS NULL OR pjh.started_at <= $2)
+ORDER BY pjh.started_at DESC
+")
+    .bind(params.from)
+    .bind(params.to)
+    .fetch_all(pool)
+    .await
+    .context("patch history query failed")?;
+
+    let mut wtr = csv::Writer::from_writer(vec![]);
+    wtr.write_record(&[
+        "job_id", "job_kind", "job_status", "host_display_name", "host_fqdn",
+        "package_count", "started_at", "completed_at", "duration_seconds", "operator",
+    ])?;
+
+    for row in &rows {
+        use sqlx::Row;
+        let job_id: String       = row.try_get("job_id").unwrap_or_default();
+        let job_kind: String     = row.try_get("job_kind").unwrap_or_default();
+        let job_status: String   = row.try_get("job_status").unwrap_or_default();
+        let display_name: String = row.try_get("display_name").unwrap_or_default();
+        let fqdn: String         = row.try_get("fqdn").unwrap_or_default();
+        let package_count: i64   = row.try_get("package_count").unwrap_or(0);
+        let started_at: Option<chrono::DateTime<chrono::Utc>> =
+            row.try_get("started_at").unwrap_or(None);
+        let completed_at: Option<chrono::DateTime<chrono::Utc>> =
+            row.try_get("completed_at").unwrap_or(None);
+        let duration_seconds: Option<i64> = row.try_get("duration_seconds").unwrap_or(None);
+        let operator: String     = row.try_get("operator").unwrap_or_default();
+
+        wtr.write_record(&[
+            job_id,
+            job_kind,
+            job_status,
+            display_name,
+            fqdn,
+            package_count.to_string(),
+            started_at.map(|d| d.to_rfc3339()).unwrap_or_default(),
+            completed_at.map(|d| d.to_rfc3339()).unwrap_or_default(),
+            duration_seconds.unwrap_or(0).to_string(),
+            operator,
+        ])?;
+    }
+
+    Ok(wtr.into_inner().context("csv flush failed")?)
+}
+
+// ---------------------------------------------------------------------------
+// Vulnerability
+// ---------------------------------------------------------------------------
+
+async fn vulnerability_csv(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    let mut wtr = csv::Writer::from_writer(vec![]);
+    wtr.write_record(&[
+        "host_id", "display_name", "fqdn", "cve_id",
+        "package_name", "severity", "available_version", "last_seen_at",
+    ])?;
+
+    let result = sqlx::query("
+SELECT
+    h.id::text        AS host_id,
+    h.display_name,
+    h.fqdn,
+    cve.cve_id,
+    cve.package_name,
+    cve.severity,
+    cve.available_version,
+    pd.updated_at     AS last_seen_at
+FROM hosts h
+JOIN host_patch_data pd ON pd.host_id = h.id
+CROSS JOIN LATERAL jsonb_to_recordset(COALESCE(pd.cve_data, '[]'::jsonb))
+  AS cve(cve_id text, package_name text, severity text, available_version text)
+WHERE ($1::timestamptz IS NULL OR pd.updated_at >= $1)
+  AND ($2::timestamptz IS NULL OR pd.updated_at <= $2)
+ORDER BY
+    CASE cve.severity
+        WHEN 'critical' THEN 1
+        WHEN 'high'     THEN 2
+        WHEN 'medium'   THEN 3
+        ELSE 4
+    END,
+    h.display_name
+")
+    .bind(params.from)
+    .bind(params.to)
+    .fetch_all(pool)
+    .await;
+
+    match result {
+        Ok(rows) => {
+            for row in &rows {
+                use sqlx::Row;
+                let host_id: String       = row.try_get("host_id").unwrap_or_default();
+                let display_name: String  = row.try_get("display_name").unwrap_or_default();
+                let fqdn: String          = row.try_get("fqdn").unwrap_or_default();
+                let cve_id: String        = row.try_get("cve_id").unwrap_or_default();
+                let package_name: String  = row.try_get("package_name").unwrap_or_default();
+                let severity: String      = row.try_get("severity").unwrap_or_default();
+                let available_version: String =
+                    row.try_get("available_version").unwrap_or_default();
+                let last_seen_at: Option<chrono::DateTime<chrono::Utc>> =
+                    row.try_get("last_seen_at").unwrap_or(None);
+
+                wtr.write_record(&[
+                    host_id,
+                    display_name,
+                    fqdn,
+                    cve_id,
+                    package_name,
+                    severity,
+                    available_version,
+                    last_seen_at.map(|d| d.to_rfc3339()).unwrap_or_default(),
+                ])?;
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "vulnerability query failed — returning empty rows");
+            // write a comment row indicating empty data
+            wtr.write_record(&[
+                "(no data)", "", "", "", "", "", "",
+                &format!("query error: {}", e),
+            ])?;
+        }
+    }
+
+    Ok(wtr.into_inner().context("csv flush failed")?)
+}
+
+// ---------------------------------------------------------------------------
+// Audit
+// ---------------------------------------------------------------------------
+
+async fn audit_csv(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    let rows = sqlx::query("
+SELECT
+    id::text            AS id,
+    created_at,
+    action::text        AS action,
+    actor_username,
+    target_type,
+    target_id,
+    ip_address::text    AS ip_address,
+    request_id
+FROM audit_log
+WHERE ($1::timestamptz IS NULL OR created_at >= $1)
+  AND ($2::timestamptz IS NULL OR created_at <= $2)
+ORDER BY created_at DESC
+LIMIT 10000
+")
+    .bind(params.from)
+    .bind(params.to)
+    .fetch_all(pool)
+    .await
+    .context("audit query failed")?;
+
+    let mut wtr = csv::Writer::from_writer(vec![]);
+    wtr.write_record(&[
+        "id", "created_at", "action", "actor_username",
+        "target_type", "target_id", "ip_address", "request_id",
+    ])?;
+
+    for row in &rows {
+        use sqlx::Row;
+        let id: String             = row.try_get("id").unwrap_or_default();
+        let created_at: Option<chrono::DateTime<chrono::Utc>> =
+            row.try_get("created_at").unwrap_or(None);
+        let action: String         = row.try_get("action").unwrap_or_default();
+        let actor_username: String = row.try_get("actor_username").unwrap_or_default();
+        let target_type: String    = row.try_get("target_type").unwrap_or_default();
+        let target_id: String      = row.try_get("target_id").unwrap_or_default();
+        let ip_address: String     = row.try_get("ip_address").unwrap_or_default();
+        let request_id: String     = row.try_get("request_id").unwrap_or_default();
+
+        wtr.write_record(&[
+            id,
+            created_at.map(|d| d.to_rfc3339()).unwrap_or_default(),
+            action,
+            actor_username,
+            target_type,
+            target_id,
+            ip_address,
+            request_id,
+        ])?;
+    }
+
+    Ok(wtr.into_inner().context("csv flush failed")?)
+}

crates/pm-reports/src/lib.rs

@@ -1,7 +1,27 @@
 //! pm-reports — CSV and PDF report generation.
 //!
 //! Uses printpdf + plotters for in-process PDF with charts.
-//!
-//! M1: Stub. Full implementation in M9.
 pub mod csv;
 pub mod pdf;
+
+pub use csv::generate_csv;
+pub use pdf::generate_pdf;
+
+/// The type of report to generate.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum ReportType {
+    Compliance,
+    PatchHistory,
+    Vulnerability,
+    Audit,
+}
+
+/// Parameters controlling report generation.
+#[derive(Debug, Clone, serde::Deserialize)]
+pub struct ReportParams {
+    pub report_type: ReportType,
+    pub from: Option<chrono::DateTime<chrono::Utc>>,
+    pub to: Option<chrono::DateTime<chrono::Utc>>,
+    pub group_id: Option<uuid::Uuid>,
+}

crates/pm-reports/src/pdf.rs

@@ -1 +1,453 @@
-//! pdf report generation stub for M9.
+//! PDF report generation for pm-reports.
+//!
+//! Uses printpdf for document structure and plotters + image for embedded charts.
+
+use crate::{ReportParams, ReportType};
+use anyhow::Context;
+use plotters::prelude::*;
+use printpdf::{
+    BuiltinFont, ColorBits, ColorSpace, Image, ImageTransform, ImageXObject,
+    IndirectFontRef, Mm, PdfDocument, PdfLayerIndex, PdfLayerReference,
+    PdfPageIndex, Px,
+};
+
+const PAGE_W: f32 = 297.0; // A4 landscape width  (mm)
+const PAGE_H: f32 = 210.0; // A4 landscape height (mm)
+const MARGIN: f32 = 10.0;
+const ROW_H: f32 = 6.0;
+const HEADER_Y_START: f32 = 190.0;
+const NEW_PAGE_THRESHOLD: f32 = 20.0;
+
+// ---------------------------------------------------------------------------
+// Public entry point
+// ---------------------------------------------------------------------------
+
+/// Generate a PDF report and return the raw bytes.
+pub async fn generate_pdf(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    match params.report_type {
+        ReportType::Compliance    => compliance_pdf(pool, params).await,
+        ReportType::PatchHistory  => patch_history_pdf(pool, params).await,
+        ReportType::Vulnerability => vulnerability_pdf(pool, params).await,
+        ReportType::Audit         => audit_pdf(pool, params).await,
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Chart helper
+// ---------------------------------------------------------------------------
+
+/// Render a bar chart to an in-memory PNG and return the raw PNG bytes.
+fn render_bar_chart(
+    labels: &[String],
+    values: &[f64],
+    title: &str,
+) -> anyhow::Result<(Vec<u8>, u32, u32)> {
+    const W: u32 = 800;
+    const H: u32 = 400;
+
+    let mut pixel_buf = vec![0u8; (W * H * 3) as usize];
+
+    {
+        let root = BitMapBackend::with_buffer(&mut pixel_buf, (W, H))
+            .into_drawing_area();
+        root.fill(&WHITE)?;
+
+        let max_val = values
+            .iter()
+            .cloned()
+            .fold(0.0_f64, f64::max)
+            .max(1.0);
+        let n = labels.len().max(1);
+
+        let mut chart = ChartBuilder::on(&root)
+            .caption(title, ("sans-serif", 20).into_font())
+            .margin(20u32)
+            .x_label_area_size(60u32)
+            .y_label_area_size(50u32)
+            .build_cartesian_2d(0..n, 0.0..max_val * 1.1)?;
+
+        chart
+            .configure_mesh()
+            .x_labels(n.min(20))
+            .x_label_formatter(&|idx| {
+                labels
+                    .get(*idx)
+                    .map(|s| {
+                        if s.len() > 12 { s[..12].to_string() } else { s.clone() }
+                    })
+                    .unwrap_or_default()
+            })
+            .y_desc("Value")
+            .draw()?;
+
+        chart.draw_series((0..n).map(|i| {
+            let v = values.get(i).copied().unwrap_or(0.0);
+            let color = if v >= 90.0 {
+                RGBColor(76, 175, 80)
+            } else if v >= 70.0 {
+                RGBColor(255, 193, 7)
+            } else {
+                RGBColor(244, 67, 54)
+            };
+            Rectangle::new([(i, 0.0), (i + 1, v)], color.filled())
+        }))?;
+
+        root.present()?;
+    }
+
+    // Return raw RGB pixels + dimensions for direct PDF embedding
+    Ok((pixel_buf, W, H))
+}
+
+// ---------------------------------------------------------------------------
+// PDF builder
+// ---------------------------------------------------------------------------
+
+struct PdfBuilder {
+    doc: printpdf::PdfDocumentReference,
+    font: IndirectFontRef,
+    font_bold: IndirectFontRef,
+    page_idx: PdfPageIndex,
+    layer_idx: PdfLayerIndex,
+    current_y: f32,
+}
+
+impl PdfBuilder {
+    fn new(title: &str) -> anyhow::Result<Self> {
+        let doc = PdfDocument::empty(title);
+        let (page_idx, layer_idx) = doc.add_page(Mm(PAGE_W), Mm(PAGE_H), "Layer 1");
+        let font      = doc.add_builtin_font(BuiltinFont::Helvetica)?;
+        let font_bold = doc.add_builtin_font(BuiltinFont::HelveticaBold)?;
+        Ok(Self {
+            doc,
+            font,
+            font_bold,
+            page_idx,
+            layer_idx,
+            current_y: HEADER_Y_START,
+        })
+    }
+
+    fn layer(&self) -> PdfLayerReference {
+        self.doc.get_page(self.page_idx).get_layer(self.layer_idx)
+    }
+
+    fn write_text(&self, s: &str, font_size: f32, x: f32, y: f32, bold: bool) {
+        let f = if bold { &self.font_bold } else { &self.font };
+        self.layer().use_text(s, font_size, Mm(x), Mm(y), f);
+    }
+
+    fn new_page(&mut self) {
+        let (pi, li) = self.doc.add_page(Mm(PAGE_W), Mm(PAGE_H), "Layer 1");
+        self.page_idx = pi;
+        self.layer_idx = li;
+        self.current_y = HEADER_Y_START;
+    }
+
+    fn ensure_space(&mut self, needed: f32) {
+        if self.current_y - needed < NEW_PAGE_THRESHOLD {
+            self.new_page();
+        }
+    }
+
+    fn table_row(&mut self, cells: &[&str], col_x: &[f32], font_size: f32, bold: bool) {
+        self.ensure_space(ROW_H);
+        let y = self.current_y;
+        for (i, cell) in cells.iter().enumerate() {
+            let x = col_x.get(i).copied().unwrap_or(MARGIN);
+            let s = if cell.len() > 30 { &cell[..30] } else { cell };
+            self.write_text(s, font_size, x, y, bold);
+        }
+        self.current_y -= ROW_H;
+    }
+
+    fn embed_image(
+        &self,
+        raw_rgb: Vec<u8>,
+        img_w: u32,
+        img_h: u32,
+        x_mm: f32,
+        y_mm: f32,
+        scale_x: f32,
+        scale_y: f32,
+    ) -> anyhow::Result<()> {
+        let xobj = ImageXObject {
+            width: Px(img_w as usize),
+            height: Px(img_h as usize),
+            color_space: ColorSpace::Rgb,
+            bits_per_component: ColorBits::Bit8,
+            interpolate: true,
+            image_data: raw_rgb,
+            image_filter: None,
+            smask: None,
+            clipping_bbox: None,
+        };
+        let pdf_img = Image::from(xobj);
+        pdf_img.add_to_layer(
+            self.layer(),
+            ImageTransform {
+                translate_x: Some(Mm(x_mm)),
+                translate_y: Some(Mm(y_mm)),
+                scale_x: Some(scale_x),
+                scale_y: Some(scale_y),
+                dpi: Some(150.0),
+                ..Default::default()
+            },
+        );
+        Ok(())
+    }
+
+    fn save(self) -> anyhow::Result<Vec<u8>> {
+        Ok(self.doc.save_to_bytes()?)
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Title page helper
+// ---------------------------------------------------------------------------
+
+fn write_title_page(pdf: &mut PdfBuilder, title: &str, params: &ReportParams) {
+    pdf.write_text(title, 24.0, MARGIN, 160.0, true);
+    pdf.write_text(
+        &format!("Generated: {}", chrono::Utc::now().format("%Y-%m-%d %H:%M UTC")),
+        11.0, MARGIN, 148.0, false,
+    );
+    if let Some(from) = params.from {
+        pdf.write_text(
+            &format!("From: {}", from.format("%Y-%m-%d")),
+            10.0, MARGIN, 140.0, false,
+        );
+    }
+    if let Some(to) = params.to {
+        pdf.write_text(
+            &format!("To:   {}", to.format("%Y-%m-%d")),
+            10.0, MARGIN, 134.0, false,
+        );
+    }
+    if let Some(gid) = params.group_id {
+        pdf.write_text(&format!("Group: {}", gid), 10.0, MARGIN, 128.0, false);
+    }
+    pdf.new_page();
+}
+
+// ---------------------------------------------------------------------------
+// Compliance PDF
+// ---------------------------------------------------------------------------
+
+async fn compliance_pdf(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    use sqlx::Row;
+    let rows = if let Some(gid) = params.group_id {
+        sqlx::query("
+SELECT h.display_name, h.fqdn,
+    COALESCE(pd.total_packages,0) AS total_packages,
+    COALESCE(pd.pending_patches,0) AS pending_patches,
+    CASE WHEN COALESCE(pd.total_packages,0)=0 THEN 100.0
+         ELSE ROUND((1.0-pd.pending_patches::float/NULLIF(pd.total_packages,0))*100,1)
+    END AS compliance_pct,
+    h.health_status::text AS health_status
+FROM hosts h LEFT JOIN host_patch_data pd ON pd.host_id=h.id
+WHERE h.id IN (SELECT host_id FROM host_groups WHERE group_id=$1)
+GROUP BY h.id,pd.total_packages,pd.pending_patches
+ORDER BY compliance_pct ASC")
+            .bind(gid).fetch_all(pool).await
+            .context("compliance PDF query (group) failed")?
+    } else {
+        sqlx::query("
+SELECT h.display_name, h.fqdn,
+    COALESCE(pd.total_packages,0) AS total_packages,
+    COALESCE(pd.pending_patches,0) AS pending_patches,
+    CASE WHEN COALESCE(pd.total_packages,0)=0 THEN 100.0
+         ELSE ROUND((1.0-pd.pending_patches::float/NULLIF(pd.total_packages,0))*100,1)
+    END AS compliance_pct,
+    h.health_status::text AS health_status
+FROM hosts h LEFT JOIN host_patch_data pd ON pd.host_id=h.id
+GROUP BY h.id,pd.total_packages,pd.pending_patches
+ORDER BY compliance_pct ASC")
+            .fetch_all(pool).await
+            .context("compliance PDF query failed")?
+    };
+    let labels: Vec<String> = rows.iter().map(|r| r.try_get::<String,_>("display_name").unwrap_or_default()).collect();
+    let values: Vec<f64>   = rows.iter().map(|r| r.try_get::<f64,_>("compliance_pct").unwrap_or(0.0)).collect();
+    let mut pdf = PdfBuilder::new("Compliance Report")?;
+    write_title_page(&mut pdf, "Compliance Report", params);
+    let col_x: &[f32] = &[MARGIN, 65.0, 130.0, 165.0, 200.0, 235.0];
+    pdf.table_row(&["Host","FQDN","Total Pkgs","Pending","Compliance %","Status"], col_x, 9.0, true);
+    for row in &rows {
+        let name:   String = row.try_get("display_name").unwrap_or_default();
+        let fqdn:   String = row.try_get("fqdn").unwrap_or_default();
+        let total:  i64    = row.try_get("total_packages").unwrap_or(0);
+        let pend:   i64    = row.try_get("pending_patches").unwrap_or(0);
+        let pct:    f64    = row.try_get("compliance_pct").unwrap_or(0.0);
+        let status: String = row.try_get("health_status").unwrap_or_default();
+        pdf.table_row(
+            &[&name,&fqdn,&total.to_string(),&pend.to_string(),&format!("{:.1}%",pct),&status],
+            col_x, 8.0, false,
+        );
+    }
+    if !labels.is_empty() {
+        match render_bar_chart(&labels, &values, "Compliance % by Host") {
+            Ok((raw, w, h)) => {
+                pdf.new_page();
+                pdf.write_text("Compliance Chart", 16.0, MARGIN, 200.0, true);
+                if let Err(e) = pdf.embed_image(raw, w, h, MARGIN, 10.0, 0.18, 0.18) {
+                    tracing::warn!(error = %e, "chart embed failed");
+                }
+            }
+            Err(e) => tracing::warn!(error = %e, "chart render failed"),
+        }
+    }
+    pdf.save()
+}
+
+// ---------------------------------------------------------------------------
+// Patch history PDF
+// ---------------------------------------------------------------------------
+
+async fn patch_history_pdf(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    use sqlx::Row;
+    let rows = sqlx::query("
+SELECT pj.kind::text AS job_kind, pj.status::text AS job_status,
+    h.display_name, h.fqdn, pjh.started_at, pjh.completed_at,
+    EXTRACT(EPOCH FROM (pjh.completed_at-pjh.started_at))::bigint AS duration_seconds,
+    COALESCE(u.username,'system') AS operator
+FROM patch_job_hosts pjh
+JOIN patch_jobs pj ON pj.id=pjh.job_id
+JOIN hosts h ON h.id=pjh.host_id
+LEFT JOIN users u ON u.id=pj.created_by_user_id
+WHERE ($1::timestamptz IS NULL OR pjh.started_at>=$1)
+  AND ($2::timestamptz IS NULL OR pjh.started_at<=$2)
+ORDER BY pjh.started_at DESC")
+        .bind(params.from).bind(params.to).fetch_all(pool).await
+        .context("patch history PDF query failed")?;
+    let mut dc: std::collections::BTreeMap<String,f64> = std::collections::BTreeMap::new();
+    for row in &rows {
+        if let Ok(Some(s)) = row.try_get::<Option<chrono::DateTime<chrono::Utc>>,_>("started_at") {
+            *dc.entry(s.format("%Y-%m-%d").to_string()).or_insert(0.0) += 1.0;
+        }
+    }
+    let cl: Vec<String> = dc.keys().cloned().collect();
+    let cv: Vec<f64>    = dc.values().cloned().collect();
+    let mut pdf = PdfBuilder::new("Patch History Report")?;
+    write_title_page(&mut pdf, "Patch History Report", params);
+    let col_x: &[f32] = &[MARGIN,45.0,80.0,115.0,155.0,200.0,245.0,270.0];
+    pdf.table_row(&["Kind","Status","Host","FQDN","Started","Completed","Dur(s)","Operator"], col_x, 9.0, true);
+    for row in &rows {
+        let kind:   String = row.try_get("job_kind").unwrap_or_default();
+        let status: String = row.try_get("job_status").unwrap_or_default();
+        let name:   String = row.try_get("display_name").unwrap_or_default();
+        let fqdn:   String = row.try_get("fqdn").unwrap_or_default();
+        let started: String = row.try_get::<Option<chrono::DateTime<chrono::Utc>>,_>("started_at")
+            .unwrap_or(None).map(|d| d.format("%Y-%m-%d %H:%M").to_string()).unwrap_or_default();
+        let completed: String = row.try_get::<Option<chrono::DateTime<chrono::Utc>>,_>("completed_at")
+            .unwrap_or(None).map(|d| d.format("%Y-%m-%d %H:%M").to_string()).unwrap_or_default();
+        let dur: i64   = row.try_get("duration_seconds").unwrap_or(0);
+        let op:  String = row.try_get("operator").unwrap_or_default();
+        pdf.table_row(&[&kind,&status,&name,&fqdn,&started,&completed,&dur.to_string(),&op], col_x, 8.0, false);
+    }
+    if !cl.is_empty() {
+        match render_bar_chart(&cl, &cv, "Jobs per Day") {
+            Ok((raw, w, h)) => {
+                pdf.new_page();
+                pdf.write_text("Patch Activity Chart", 16.0, MARGIN, 200.0, true);
+                if let Err(e) = pdf.embed_image(raw, w, h, MARGIN, 10.0, 0.18, 0.18) {
+                    tracing::warn!(error = %e, "chart embed failed");
+                }
+            }
+            Err(e) => tracing::warn!(error = %e, "chart render failed"),
+        }
+    }
+    pdf.save()
+}
+
+// ---------------------------------------------------------------------------
+// Vulnerability PDF
+// ---------------------------------------------------------------------------
+
+async fn vulnerability_pdf(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    use sqlx::Row;
+    // Query DB FIRST (before creating any non-Send PdfBuilder)
+    let query_result = sqlx::query("
+SELECT h.display_name, h.fqdn,
+    cve.cve_id, cve.package_name, cve.severity, cve.available_version,
+    pd.updated_at AS last_seen_at
+FROM hosts h JOIN host_patch_data pd ON pd.host_id=h.id
+CROSS JOIN LATERAL jsonb_to_recordset(COALESCE(pd.cve_data,'[]'::jsonb))
+  AS cve(cve_id text,package_name text,severity text,available_version text)
+WHERE ($1::timestamptz IS NULL OR pd.updated_at>=$1)
+  AND ($2::timestamptz IS NULL OR pd.updated_at<=$2)
+ORDER BY CASE cve.severity WHEN 'critical' THEN 1 WHEN 'high' THEN 2 WHEN 'medium' THEN 3 ELSE 4 END,
+    h.display_name")
+        .bind(params.from).bind(params.to).fetch_all(pool).await;
+    // Now create PdfBuilder (non-Send Rc types) after all awaits
+    let mut pdf = PdfBuilder::new("Vulnerability Report")?;
+    write_title_page(&mut pdf, "Vulnerability Exposure Report", params);
+    let col_x: &[f32] = &[MARGIN,55.0,100.0,130.0,175.0,215.0,255.0];
+    pdf.table_row(&["Host","FQDN","CVE ID","Package","Severity","Fix Version","Last Seen"], col_x, 9.0, true);
+    match query_result {
+        Ok(rows) => {
+            for row in &rows {
+                let name: String = row.try_get("display_name").unwrap_or_default();
+                let fqdn: String = row.try_get("fqdn").unwrap_or_default();
+                let cve:  String = row.try_get("cve_id").unwrap_or_default();
+                let pkg:  String = row.try_get("package_name").unwrap_or_default();
+                let sev:  String = row.try_get("severity").unwrap_or_default();
+                let fix:  String = row.try_get("available_version").unwrap_or_default();
+                let seen: String = row.try_get::<Option<chrono::DateTime<chrono::Utc>>,_>("last_seen_at")
+                    .unwrap_or(None).map(|d| d.format("%Y-%m-%d").to_string()).unwrap_or_default();
+                pdf.table_row(&[&name,&fqdn,&cve,&pkg,&sev,&fix,&seen], col_x, 8.0, false);
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "vulnerability PDF query failed");
+            let y = pdf.current_y;
+            pdf.write_text(&format!("No data: {}", e), 10.0, MARGIN, y, false);
+        }
+    }
+    pdf.save()
+}
+
+
+async fn audit_pdf(
+    pool: &sqlx::PgPool,
+    params: &ReportParams,
+) -> anyhow::Result<Vec<u8>> {
+    use sqlx::Row;
+    let rows = sqlx::query("
+SELECT id::text AS id, created_at, action::text AS action,
+    actor_username, target_type, target_id,
+    ip_address::text AS ip_address, request_id
+FROM audit_log
+WHERE ($1::timestamptz IS NULL OR created_at>=$1)
+  AND ($2::timestamptz IS NULL OR created_at<=$2)
+ORDER BY created_at DESC LIMIT 10000")
+        .bind(params.from).bind(params.to).fetch_all(pool).await
+        .context("audit PDF query failed")?;
+    let mut pdf = PdfBuilder::new("Audit Trail Report")?;
+    write_title_page(&mut pdf, "Audit Trail Report", params);
+    let col_x: &[f32] = &[MARGIN,50.0,95.0,135.0,175.0,215.0,255.0];
+    pdf.table_row(&["Timestamp","Action","Actor","Target Type","Target ID","IP","Request ID"], col_x, 9.0, true);
+    for row in &rows {
+        let created: String = row.try_get::<Option<chrono::DateTime<chrono::Utc>>,_>("created_at")
+            .unwrap_or(None).map(|d| d.format("%Y-%m-%d %H:%M").to_string()).unwrap_or_default();
+        let action: String = row.try_get("action").unwrap_or_default();
+        let actor:  String = row.try_get("actor_username").unwrap_or_default();
+        let ttype:  String = row.try_get("target_type").unwrap_or_default();
+        let tid:    String = row.try_get("target_id").unwrap_or_default();
+        let ip:     String = row.try_get("ip_address").unwrap_or_default();
+        let req:    String = row.try_get("request_id").unwrap_or_default();
+        pdf.table_row(&[&created,&action,&actor,&ttype,&tid,&ip,&req], col_x, 8.0, false);
+    }
+    pdf.save()
+}

crates/pm-web/Cargo.toml

@@ -10,8 +10,10 @@ name = "pm-web"
 path = "src/main.rs"
 
 [dependencies]
+pm-ca = { path = "../pm-ca" }
 pm-core = { path = "../pm-core" }
 pm-auth = { path = "../pm-auth" }
+pm-reports = { path = "../pm-reports" }
 tokio = { workspace = true }
 axum = { workspace = true }
 axum-extra = { workspace = true }

crates/pm-web/src/main.rs

@@ -42,6 +42,8 @@ pub struct AppState {
     pub auth_config: Arc<AuthConfig>,
     /// In-memory store for single-use WebSocket authentication tickets.
     pub ws_tickets: Arc<DashMap<String, WsTicket>>,
+    /// Internal certificate authority for mTLS client cert issuance.
+    pub ca: Arc<pm_ca::CertAuthority>,
 }
 
 #[tokio::main]
@@ -77,6 +79,16 @@ async fn main() -> anyhow::Result<()> {
     let pool = db::init_pool(&config.database).await?;
     db::run_migrations(&pool).await?;
 
+    // Initialise the internal CA.  Panics in production if CA files are missing
+    // or corrupt — this is intentional; the service cannot operate without mTLS.
+    let ca_base = std::path::Path::new("/etc/patch-manager/ca");
+    let ca = pm_ca::CertAuthority::init(ca_base, &pool)
+        .await
+        .unwrap_or_else(|e| {
+            tracing::warn!(error = %e, "CA init failed (dev mode)");
+            panic!("CA initialization failed: {}", e);
+        });
+
     let ws_tickets: Arc<DashMap<String, WsTicket>> = Arc::new(DashMap::new());
 
     // Background task: purge expired WS tickets every 30 seconds.
@@ -103,6 +115,7 @@ async fn main() -> anyhow::Result<()> {
         signing_key_pem,
         auth_config,
         ws_tickets,
+        ca: Arc::new(ca),
     };
 
     let app = build_router(state);
@@ -128,6 +141,8 @@ pub fn build_router(state: AppState) -> Router {
         .merge(routes::auth::protected_router())
         // Hosts
         .nest("/hosts", routes::hosts::router())
+        // Host-scoped certificate endpoints (merged separately to avoid conflict)
+        .nest("/hosts", routes::ca::host_cert_router())
         // Groups
         .nest("/groups", routes::groups::router())
         // Users
@@ -140,8 +155,14 @@ pub fn build_router(state: AppState) -> Router {
         .nest("/jobs", routes::jobs::router())
         // Maintenance windows (nested under hosts path param)
         .nest("/hosts/:host_id/maintenance-windows", routes::maintenance_windows::router())
+        // CA root certificate download
+        .nest("/ca", routes::ca::ca_router())
+        // Certificate list / renew / revoke
+        .nest("/certificates", routes::ca::certs_router())
         // WS ticket issuance (JWT-protected — ticket returned to browser, then used for WS upgrade)
         .merge(routes::ws::ticket_router())
+        // Reports
+        .nest("/reports", routes::reports::router())
         // Apply auth middleware to all the above
         .route_layer(middleware::from_fn(move |req, next| {
             let auth_config = auth_config.clone();

crates/pm-web/src/routes/ca.rs

@@ -0,0 +1,349 @@
+//! CA / certificate management routes.
+//!
+//! ca_router()        → mounted at /api/v1/ca
+//!   GET  /root.crt                           download_root_ca      (any authed role)
+//!
+//! certs_router()     → mounted at /api/v1/certificates
+//!   GET  /                                   list_certificates     (any authed role)
+//!   POST /:cert_id/renew                     renew_cert            (admin only)
+//!   DELETE /:cert_id                         revoke_cert           (admin only)
+//!
+//! host_cert_router() → merged under /api/v1/hosts
+//!   GET  /:host_id/client.crt                download_client_cert  (admin only)
+//!   POST /:host_id/certificates              issue_client_cert     (admin only)
+
+use axum::{
+    body::Body,
+    extract::{Path, Query, State},
+    http::{header, Response, StatusCode},
+    response::Json,
+    routing::{delete, get, post},
+    Router,
+};
+use chrono::{DateTime, Utc};
+use pm_auth::rbac::AuthUser;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use uuid::Uuid;
+
+use crate::AppState;
+
+// ── Router constructors ───────────────────────────────────────────────────────
+
+/// Handles routes mounted at /api/v1/ca
+pub fn ca_router() -> Router<AppState> {
+    Router::new()
+        .route("/root.crt", get(download_root_ca))
+}
+
+/// Handles routes mounted at /api/v1/certificates
+pub fn certs_router() -> Router<AppState> {
+    Router::new()
+        .route("/", get(list_certificates))
+        .route("/:cert_id/renew", post(renew_cert))
+        .route("/:cert_id", delete(revoke_cert))
+}
+
+/// Handles cert-specific paths merged under /api/v1/hosts.
+/// Only adds paths not already claimed by the hosts router.
+pub fn host_cert_router() -> Router<AppState> {
+    Router::new()
+        .route("/:host_id/client.crt", get(download_client_cert))
+        .route("/:host_id/certificates", post(issue_client_cert))
+}
+
+// ── Shared types ──────────────────────────────────────────────────────────────
+
+/// Row returned from the `certificates` table.
+#[derive(Debug, Serialize, sqlx::FromRow)]
+struct CertRow {
+    id: Uuid,
+    host_id: Option<Uuid>,
+    serial_number: String,
+    common_name: String,
+    /// Cast to TEXT in all queries to avoid custom-enum decode.
+    status: String,
+    issued_at: DateTime<Utc>,
+    expires_at: DateTime<Utc>,
+    revoked_at: Option<DateTime<Utc>>,
+}
+
+/// Query params for `list_certificates`.
+#[derive(Debug, Deserialize)]
+struct CertListQuery {
+    host_id: Option<Uuid>,
+    status: Option<String>,
+}
+
+/// Request body for `issue_client_cert`.
+#[derive(Debug, Deserialize)]
+struct IssueCertRequest {
+    hostname: String,
+}
+
+// ── Helper: build PEM download response ──────────────────────────────────────
+
+fn pem_response(
+    pem: String,
+    filename: &str,
+) -> Result<Response<Body>, (StatusCode, Json<Value>)> {
+    let disposition = format!("attachment; filename=\"{filename}\"");
+    Response::builder()
+        .status(StatusCode::OK)
+        .header(header::CONTENT_TYPE, "application/x-pem-file")
+        .header(header::CONTENT_DISPOSITION, disposition)
+        .body(Body::from(pem))
+        .map_err(|e| {
+            tracing::error!(error = %e, "Failed to build PEM response");
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({ "error": { "code": "internal_error", "message": "Response build error" } })),
+            )
+        })
+}
+
+// ── Helper: admin-only guard ──────────────────────────────────────────────────
+
+fn require_admin(user: &AuthUser) -> Result<(), (StatusCode, Json<Value>)> {
+    if !user.role.is_admin() {
+        return Err((
+            StatusCode::FORBIDDEN,
+            Json(json!({ "error": { "code": "forbidden", "message": "Admin role required" } })),
+        ));
+    }
+    Ok(())
+}
+
+// ── Helper: map sqlx error to 500 ─────────────────────────────────────────────
+
+fn db_error(e: sqlx::Error) -> (StatusCode, Json<Value>) {
+    tracing::error!(error = %e, "Database error");
+    (
+        StatusCode::INTERNAL_SERVER_ERROR,
+        Json(json!({ "error": { "code": "internal_error", "message": "Database error" } })),
+    )
+}
+
+// ── GET /api/v1/ca/root.crt ───────────────────────────────────────────────────
+
+/// Download the root CA certificate as a PEM file.
+async fn download_root_ca(
+    State(state): State<AppState>,
+    _auth: AuthUser,
+) -> Result<Response<Body>, (StatusCode, Json<Value>)> {
+    let pem = state.ca.root_cert_pem().to_owned();
+    pem_response(pem, "ca.crt")
+}
+
+// ── GET /api/v1/certificates ──────────────────────────────────────────────────
+
+/// List certificates with optional `?host_id=` and `?status=` filters.
+async fn list_certificates(
+    State(state): State<AppState>,
+    _auth: AuthUser,
+    Query(q): Query<CertListQuery>,
+) -> Result<Json<Vec<CertRow>>, (StatusCode, Json<Value>)> {
+    // Use the non-macro query_as form — avoids needing DATABASE_URL at compile
+    // time.  status is cast to TEXT so sqlx decodes it into String directly.
+    let rows: Vec<CertRow> = match (q.host_id, q.status.as_deref()) {
+        (Some(hid), Some(st)) => {
+            sqlx::query_as::<_, CertRow>(
+                r#"SELECT id, host_id, serial_number, common_name,
+                          status::text AS status,
+                          issued_at, expires_at, revoked_at
+                   FROM certificates
+                   WHERE host_id = $1 AND status::text = $2
+                   ORDER BY issued_at DESC"#,
+            )
+            .bind(hid)
+            .bind(st)
+            .fetch_all(&state.db)
+            .await
+        }
+        (Some(hid), None) => {
+            sqlx::query_as::<_, CertRow>(
+                r#"SELECT id, host_id, serial_number, common_name,
+                          status::text AS status,
+                          issued_at, expires_at, revoked_at
+                   FROM certificates
+                   WHERE host_id = $1
+                   ORDER BY issued_at DESC"#,
+            )
+            .bind(hid)
+            .fetch_all(&state.db)
+            .await
+        }
+        (None, Some(st)) => {
+            sqlx::query_as::<_, CertRow>(
+                r#"SELECT id, host_id, serial_number, common_name,
+                          status::text AS status,
+                          issued_at, expires_at, revoked_at
+                   FROM certificates
+                   WHERE status::text = $1
+                   ORDER BY issued_at DESC"#,
+            )
+            .bind(st)
+            .fetch_all(&state.db)
+            .await
+        }
+        (None, None) => {
+            sqlx::query_as::<_, CertRow>(
+                r#"SELECT id, host_id, serial_number, common_name,
+                          status::text AS status,
+                          issued_at, expires_at, revoked_at
+                   FROM certificates
+                   ORDER BY issued_at DESC"#,
+            )
+            .fetch_all(&state.db)
+            .await
+        }
+    }
+    .map_err(db_error)?;
+
+    Ok(Json(rows))
+}
+
+// ── GET /api/v1/hosts/:host_id/client.crt ────────────────────────────────────
+
+/// Download the most recent active client certificate PEM for a host.
+async fn download_client_cert(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(host_id): Path<Uuid>,
+) -> Result<Response<Body>, (StatusCode, Json<Value>)> {
+    require_admin(&auth)?;
+
+    let cert_pem: Option<String> = sqlx::query_scalar(
+        r#"SELECT cert_pem
+           FROM certificates
+           WHERE host_id = $1
+             AND status = 'active'::cert_status
+           ORDER BY issued_at DESC
+           LIMIT 1"#,
+    )
+    .bind(host_id)
+    .fetch_optional(&state.db)
+    .await
+    .map_err(|e| {
+        tracing::error!(error = %e, %host_id, "Failed to fetch client cert");
+        db_error(e)
+    })?;
+
+    match cert_pem {
+        Some(pem) => pem_response(pem, "client.crt"),
+        None => Err((
+            StatusCode::NOT_FOUND,
+            Json(json!({
+                "error": {
+                    "code": "not_found",
+                    "message": "No active certificate found for this host"
+                }
+            })),
+        )),
+    }
+}
+
+// ── POST /api/v1/hosts/:host_id/certificates ─────────────────────────────────
+
+/// Issue a new mTLS client certificate for a host.
+/// **The private key is returned only once — the caller must save it.**
+async fn issue_client_cert(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(host_id): Path<Uuid>,
+    Json(req): Json<IssueCertRequest>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    require_admin(&auth)?;
+
+    let issued = state
+        .ca
+        .issue_client_cert(host_id, &req.hostname, &state.db)
+        .await
+        .map_err(|e| {
+            tracing::error!(error = %e, %host_id, hostname = %req.hostname,
+                "Failed to issue client cert");
+            (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({ "error": { "code": "internal_error", "message": e.to_string() } })),
+            )
+        })?;
+
+    Ok(Json(json!({
+        "cert_pem":      issued.cert_pem,
+        "key_pem":       issued.key_pem,
+        "serial_number": issued.serial_number,
+        "expires_at":    issued.expires_at,
+    })))
+}
+
+// ── POST /api/v1/certificates/:cert_id/renew ─────────────────────────────────
+
+/// Revoke the specified certificate and issue a replacement with the same CN.
+async fn renew_cert(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(cert_id): Path<Uuid>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    require_admin(&auth)?;
+
+    let issued = state
+        .ca
+        .renew_cert(cert_id, &state.db)
+        .await
+        .map_err(|e| {
+            let msg = e.to_string();
+            tracing::error!(error = %e, %cert_id, "Failed to renew cert");
+            if msg.contains("not found") {
+                (
+                    StatusCode::NOT_FOUND,
+                    Json(json!({ "error": { "code": "not_found", "message": "Certificate not found" } })),
+                )
+            } else {
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({ "error": { "code": "internal_error", "message": msg } })),
+                )
+            }
+        })?;
+
+    Ok(Json(json!({
+        "cert_pem":      issued.cert_pem,
+        "key_pem":       issued.key_pem,
+        "serial_number": issued.serial_number,
+        "expires_at":    issued.expires_at,
+    })))
+}
+
+// ── DELETE /api/v1/certificates/:cert_id ─────────────────────────────────────
+
+/// Revoke a certificate by ID. Sets status to 'revoked' in the database.
+async fn revoke_cert(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(cert_id): Path<Uuid>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    require_admin(&auth)?;
+
+    state
+        .ca
+        .revoke_cert(cert_id, &state.db)
+        .await
+        .map_err(|e| {
+            let msg = e.to_string();
+            tracing::error!(error = %e, %cert_id, "Failed to revoke cert");
+            if msg.contains("not found") {
+                (
+                    StatusCode::NOT_FOUND,
+                    Json(json!({ "error": { "code": "not_found", "message": "Certificate not found" } })),
+                )
+            } else {
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({ "error": { "code": "internal_error", "message": msg } })),
+                )
+            }
+        })?;
+
+    tracing::info!(%cert_id, "Certificate revoked via API");
+    Ok(Json(json!({ "revoked": true })))
+}

crates/pm-web/src/routes/mod.rs

@@ -1,5 +1,6 @@
 //! Route modules for the pm-web API.
 pub mod auth;
+pub mod ca;
 pub mod discovery;
 pub mod groups;
 pub mod hosts;
@@ -8,3 +9,5 @@ pub mod jobs;
 pub mod status;
 pub mod users;
 pub mod ws;
+
+pub mod reports;

crates/pm-web/src/routes/reports.rs

@@ -0,0 +1,153 @@
+//! Report generation endpoints.
+//!
+//! GET /api/v1/reports/compliance?format=csv|pdf&from=...&to=...&group_id=...
+//! GET /api/v1/reports/patch-history?format=csv|pdf&from=...&to=...
+//! GET /api/v1/reports/vulnerability?format=csv|pdf&from=...&to=...
+//! GET /api/v1/reports/audit?format=csv|pdf&from=...&to=...
+
+use axum::{
+    body::Bytes,
+    extract::{Query, State},
+    http::{header, HeaderMap, HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+    routing::get,
+    Router,
+};
+use pm_reports::{ReportParams, ReportType};
+
+use crate::AppState;
+
+#[derive(serde::Deserialize)]
+struct ReportQuery {
+    /// "csv" or "pdf" (defaults to "csv")
+    format: Option<String>,
+    from: Option<chrono::DateTime<chrono::Utc>>,
+    to: Option<chrono::DateTime<chrono::Utc>>,
+    group_id: Option<uuid::Uuid>,
+}
+
+pub fn router() -> Router<AppState> {
+    Router::new()
+        .route("/compliance",    get(compliance_report))
+        .route("/patch-history", get(patch_history_report))
+        .route("/vulnerability", get(vulnerability_report))
+        .route("/audit",         get(audit_report))
+}
+
+// ---------------------------------------------------------------------------
+// Internal helper
+// ---------------------------------------------------------------------------
+
+async fn run_report(
+    db: sqlx::PgPool,
+    params: ReportParams,
+    use_pdf: bool,
+    csv_name: &'static str,
+    pdf_name: &'static str,
+) -> Response {
+    let (ct, disposition, result) = if use_pdf {
+        let disp = format!("attachment; filename=\"{}\"", pdf_name);
+        let data = pm_reports::generate_pdf(&db, &params).await;
+        ("application/pdf", disp, data)
+    } else {
+        let disp = format!("attachment; filename=\"{}\"", csv_name);
+        let data = pm_reports::generate_csv(&db, &params).await;
+        ("text/csv; charset=utf-8", disp, data)
+    };
+
+    match result {
+        Ok(bytes) => {
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                header::CONTENT_TYPE,
+                HeaderValue::from_static(ct),
+            );
+            headers.insert(
+                header::CONTENT_DISPOSITION,
+                HeaderValue::from_str(&disposition)
+                    .unwrap_or_else(|_| HeaderValue::from_static("attachment")),
+            );
+            (headers, Bytes::from(bytes)).into_response()
+        }
+        Err(e) => {
+            tracing::error!(error = %e, "report generation failed");
+            (StatusCode::INTERNAL_SERVER_ERROR, format!("Report error: {}", e)).into_response()
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Handlers
+// ---------------------------------------------------------------------------
+
+async fn compliance_report(
+    State(state): State<AppState>,
+    Query(q): Query<ReportQuery>,
+) -> Response {
+    let params = ReportParams {
+        report_type: ReportType::Compliance,
+        from: q.from,
+        to: q.to,
+        group_id: q.group_id,
+    };
+    let use_pdf = matches!(q.format.as_deref(), Some("pdf"));
+    run_report(
+        state.db, params, use_pdf,
+        "compliance-report.csv",
+        "compliance-report.pdf",
+    ).await
+}
+
+async fn patch_history_report(
+    State(state): State<AppState>,
+    Query(q): Query<ReportQuery>,
+) -> Response {
+    let params = ReportParams {
+        report_type: ReportType::PatchHistory,
+        from: q.from,
+        to: q.to,
+        group_id: q.group_id,
+    };
+    let use_pdf = matches!(q.format.as_deref(), Some("pdf"));
+    run_report(
+        state.db, params, use_pdf,
+        "patch-history-report.csv",
+        "patch-history-report.pdf",
+    ).await
+}
+
+async fn vulnerability_report(
+    State(state): State<AppState>,
+    Query(q): Query<ReportQuery>,
+) -> Response {
+    let params = ReportParams {
+        report_type: ReportType::Vulnerability,
+        from: q.from,
+        to: q.to,
+        group_id: q.group_id,
+    };
+    let use_pdf = matches!(q.format.as_deref(), Some("pdf"));
+    run_report(
+        state.db, params, use_pdf,
+        "vulnerability-report.csv",
+        "vulnerability-report.pdf",
+    ).await
+}
+
+async fn audit_report(
+    State(state): State<AppState>,
+    Query(q): Query<ReportQuery>,
+) -> Response {
+    let params = ReportParams {
+        report_type: ReportType::Audit,
+        from: q.from,
+        to: q.to,
+        group_id: q.group_id,
+    };
+    let use_pdf = matches!(q.format.as_deref(), Some("pdf"));
+    run_report(
+        state.db, params, use_pdf,
+        "audit-report.csv",
+        "audit-report.pdf",
+    ).await
+}