diff --git a/Cargo.lock b/Cargo.lock index d33fbb6..aef91e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -723,6 +723,7 @@ dependencies = [ "sha2", "thiserror 2.0.18", "tokio", + "toml", "tracing", "tracing-opentelemetry", "tracing-subscriber", diff --git a/compliance-core/Cargo.toml b/compliance-core/Cargo.toml index e1b5c50..637ede1 100644 --- a/compliance-core/Cargo.toml +++ b/compliance-core/Cargo.toml @@ -50,3 +50,7 @@ axum = { version = "0.8", optional = true } jsonwebtoken = { version = "9", optional = true } reqwest = { workspace = true, optional = true } tokio = { workspace = true, optional = true } + +[dev-dependencies] +# Parse the declarative TOML job specs in the Werkbank contract tests. +toml = "0.8" diff --git a/compliance-core/src/models/mod.rs b/compliance-core/src/models/mod.rs index a56241e..51c0755 100644 --- a/compliance-core/src/models/mod.rs +++ b/compliance-core/src/models/mod.rs @@ -15,6 +15,7 @@ pub mod repository; pub mod sbom; pub mod scan; pub(crate) mod serde_helpers; +pub mod werkbank; pub use auth::AuthInfo; pub use chat::{ChatMessage, ChatRequest, ChatResponse, SourceReference}; @@ -47,3 +48,6 @@ pub use pentest::{ pub use repository::ScanTrigger; pub use sbom::{SbomEntry, VulnRef}; pub use scan::{ScanPhase, ScanRun, ScanRunStatus, ScanType}; +pub use werkbank::{ + DastCollect, Executor, InputRef, Job, JobCollect, JobResult, JobRuntime, JobStatus, JobType, +}; diff --git a/compliance-core/src/models/werkbank.rs b/compliance-core/src/models/werkbank.rs new file mode 100644 index 0000000..5969bfd --- /dev/null +++ b/compliance-core/src/models/werkbank.rs @@ -0,0 +1,378 @@ +//! The Werkbank job/result contract (WB-01). +//! +//! The shared, dependency-free vocabulary the control plane and the Werkbank +//! execution runner agree on: what a [`Job`] is, which [`Executor`] runs it, how +//! it moves through the queue ([`JobStatus`]), and what a [`JobResult`] carries +//! back. Jobs are declarative — TOML on disk, JSON on the wire — and results +//! reuse the existing scanner result types ([`Finding`], [`DastFinding`], +//! [`SbomEntry`]) so the runner produces exactly what the control plane persists. +//! +//! This module is intentionally free of the `mongodb`/`axum` features so the +//! runner can depend on `compliance-core` without pulling the server stack. + +use std::collections::BTreeMap; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +use super::dast::DastFinding; +use super::finding::Finding; +use super::sbom::SbomEntry; + +/// The kind of dynamic-execution job. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum JobType { + /// Instantiate control logic on an ephemeral soft-PLC and probe it. + PlcProvision, + /// Boot a firmware image under QEMU and run dynamic checks. + QemuBoot, + /// Crawl and dynamically test a running web endpoint. + Dast, + /// Run an active penetration test against a running target. + Pentest, +} + +/// How a runner executes a job — the CI-runner-style classification. A runner +/// advertises exactly one; a job requires one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Executor { + /// A subprocess on the runner host (dev / trusted single-node). + Shell, + /// One or more containers on the runner's Docker (default; QEMU runs here). + Docker, + /// A Pod/Job in a Kubernetes cluster (scale-out / multi-tenant). + K8s, +} + +/// Lifecycle state of a job in the queue. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum JobStatus { + /// Waiting to be leased. + Queued, + /// Leased by a runner but not yet started. + Leased, + /// Executing on a runner. + Running, + /// Completed successfully. + Succeeded, + /// Completed with an error. + Failed, + /// The lease/lifetime deadline elapsed before completion. + Expired, + /// Cancelled by the control plane. + Cancelled, +} + +impl JobStatus { + /// Whether the job has reached a terminal state (no further transitions). + pub fn is_terminal(self) -> bool { + matches!( + self, + JobStatus::Succeeded | JobStatus::Failed | JobStatus::Expired | JobStatus::Cancelled + ) + } +} + +/// A reference to an input artifact. Resolved by the runner from a source it can +/// reach; the blob itself never flows through the control plane (so an on-prem +/// runner keeps customer data local). Exactly one of `blob`/`url` should be set. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct InputRef { + /// Content-addressed blob (e.g. `sha256:…`) the runner fetches from its store. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub blob: Option, + /// A URL the runner can reach (git repo, internal artifact store, …). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub url: Option, +} + +impl InputRef { + /// A content-addressed blob reference. + pub fn blob(id: impl Into) -> Self { + Self { + blob: Some(id.into()), + url: None, + } + } +} + +/// Sandbox runtime knobs. Fields are executor/job-type specific and all optional; +/// `extra` carries anything not modelled explicitly. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct JobRuntime { + /// Container image (Docker executor). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub image: Option, + /// Memory cap (e.g. `512m`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory: Option, + /// CPU cap (e.g. `0.5`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cpus: Option, + /// Network to join (e.g. `isolated`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub network: Option, + /// QEMU machine type (qemu-boot). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub machine: Option, + /// QEMU target architecture (qemu-boot). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub arch: Option, + /// Executor-specific extras not modelled above. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub extra: BTreeMap, +} + +/// DAST collection settings for jobs that scan a web endpoint. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct DastCollect { + /// Maximum crawl depth (kept shallow for ephemeral instances). + pub max_crawl_depth: u32, +} + +/// What to collect from a run. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct JobCollect { + /// Run the industrial-protocol probe (Modbus/OPC-UA/EtherNet-IP). + #[serde(default)] + pub ics_probe: bool, + /// Run DAST against the provisioned/booted web endpoint. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dast: Option, + /// Run an active pentest. + #[serde(default)] + pub pentest: bool, + /// Collect an SBOM. + #[serde(default)] + pub sbom: bool, +} + +/// A declarative dynamic-execution job the control plane enqueues and a Werkbank +/// runner leases and executes. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Job { + /// Unique job id (assigned by the control plane on enqueue). + pub id: String, + /// What kind of job this is. + #[serde(rename = "type")] + pub job_type: JobType, + /// Owning tenant. + pub tenant: String, + /// The onboarded target this job tests. + pub target_id: String, + /// The executor a runner must provide to run this job. + pub executor: Executor, + /// Runner capabilities this job requires (e.g. `arch=amd64`, `kvm=true`). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub labels: Vec, + /// Hard lifetime deadline for the whole job. + pub timeout_secs: u64, + /// Named input artifacts (e.g. `program`, `firmware`), by reference. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub inputs: BTreeMap, + /// Sandbox runtime knobs. + #[serde(default)] + pub runtime: JobRuntime, + /// What to collect from the run. + #[serde(default)] + pub collect: JobCollect, +} + +impl Job { + /// A `plc-provision` job: instantiate the control logic named `program` on an + /// ephemeral soft-PLC (Docker executor) and collect the ICS probe + DAST. + pub fn plc_provision( + id: impl Into, + tenant: impl Into, + target_id: impl Into, + program: InputRef, + timeout_secs: u64, + ) -> Self { + let mut inputs = BTreeMap::new(); + inputs.insert("program".to_string(), program); + Self { + id: id.into(), + job_type: JobType::PlcProvision, + tenant: tenant.into(), + target_id: target_id.into(), + executor: Executor::Docker, + labels: Vec::new(), + timeout_secs, + inputs, + runtime: JobRuntime::default(), + collect: JobCollect { + ics_probe: true, + dast: Some(DastCollect { max_crawl_depth: 2 }), + pentest: false, + sbom: false, + }, + } + } +} + +/// The outcome of running a job, posted back to the control plane. Findings and +/// SBOM reuse the shared scanner types, so the control plane persists them +/// unchanged. Submission is idempotent — keyed by [`JobResult::job_id`]. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct JobResult { + /// The job this result is for. + pub job_id: String, + /// Terminal status of the job. + pub status: Option, + /// General scanner findings (e.g. ICS-probe findings). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub findings: Vec, + /// DAST findings from a web-endpoint scan. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub dast_findings: Vec, + /// SBOM components collected from the run. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub sbom: Vec, + /// Error message when the job failed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Captured execution log (truncated by the runner). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub logs: Option, + /// When execution started on the runner. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option>, + /// When execution finished. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option>, +} + +impl JobResult { + /// A successful result for a job. + pub fn succeeded(job_id: impl Into) -> Self { + Self { + job_id: job_id.into(), + status: Some(JobStatus::Succeeded), + ..Default::default() + } + } + + /// A failed result carrying an error message. + pub fn failed(job_id: impl Into, error: impl Into) -> Self { + Self { + job_id: job_id.into(), + status: Some(JobStatus::Failed), + error: Some(error.into()), + ..Default::default() + } + } +} + +#[cfg(test)] +#[allow(clippy::expect_used, clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn job_round_trips_through_json() { + let job = Job::plc_provision("job_1", "acme", "64f0aa", InputRef::blob("sha256:abc"), 180); + let json = serde_json::to_string(&job).expect("serialize"); + let back: Job = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(job, back); + // Enum wire forms are the kebab/lowercase the contract documents. + assert!(json.contains("\"type\":\"plc-provision\"")); + assert!(json.contains("\"executor\":\"docker\"")); + } + + #[test] + fn parses_the_design_doc_plc_provision_toml() { + // The exact shape from docs/DESIGN.md §5 (wrapped in a [job] table). + #[derive(Deserialize)] + struct JobFile { + job: Job, + } + let src = r#" + [job] + id = "job_01H" + type = "plc-provision" + tenant = "acme" + target_id = "64f0" + executor = "docker" + labels = ["arch=amd64"] + timeout_secs = 180 + + [job.inputs] + program = { blob = "sha256:deadbeef" } + + [job.runtime] + image = "openplc:latest" + memory = "512m" + cpus = "0.5" + network = "isolated" + + [job.collect] + ics_probe = true + dast = { max_crawl_depth = 2 } + "#; + let file: JobFile = toml::from_str(src).expect("parse job toml"); + let job = file.job; + assert_eq!(job.job_type, JobType::PlcProvision); + assert_eq!(job.executor, Executor::Docker); + assert_eq!(job.labels, vec!["arch=amd64".to_string()]); + assert_eq!( + job.inputs.get("program").and_then(|i| i.blob.as_deref()), + Some("sha256:deadbeef") + ); + assert_eq!(job.runtime.image.as_deref(), Some("openplc:latest")); + assert!(job.collect.ics_probe); + assert_eq!(job.collect.dast.map(|d| d.max_crawl_depth), Some(2)); + } + + #[test] + fn qemu_boot_runtime_fields_parse() { + #[derive(Deserialize)] + struct JobFile { + job: Job, + } + let src = r#" + [job] + id = "j2" + type = "qemu-boot" + tenant = "acme" + target_id = "t" + executor = "docker" + labels = ["kvm=true"] + timeout_secs = 600 + [job.inputs] + firmware = { blob = "sha256:cafe" } + [job.runtime] + machine = "virt" + arch = "arm" + memory = "1g" + "#; + let file: JobFile = toml::from_str(src).expect("parse"); + assert_eq!(file.job.job_type, JobType::QemuBoot); + assert_eq!(file.job.runtime.arch.as_deref(), Some("arm")); + assert_eq!( + file.job + .inputs + .get("firmware") + .and_then(|i| i.blob.as_deref()), + Some("sha256:cafe") + ); + } + + #[test] + fn status_terminality() { + assert!(JobStatus::Succeeded.is_terminal()); + assert!(JobStatus::Expired.is_terminal()); + assert!(!JobStatus::Queued.is_terminal()); + assert!(!JobStatus::Running.is_terminal()); + } + + #[test] + fn result_constructors() { + assert_eq!(JobResult::succeeded("j").status, Some(JobStatus::Succeeded)); + let f = JobResult::failed("j", "boom"); + assert_eq!(f.status, Some(JobStatus::Failed)); + assert_eq!(f.error.as_deref(), Some("boom")); + } +}