diff --git a/backend/controlplane/src/integrations/mod.rs b/backend/controlplane/src/integrations/mod.rs new file mode 100644 index 0000000..632982f --- /dev/null +++ b/backend/controlplane/src/integrations/mod.rs @@ -0,0 +1,11 @@ +//! Enterprise Integrations — governed connectors to enterprise & government systems. + +pub mod model; +pub mod placeholders; +pub mod store; + +pub use model::{ + classify_risk, CredentialRef, IntegrationAuditEvent, IntegrationKind, IntegrationPermission, + IntegrationProvider, NewIntegration, +}; +pub use store::IntegrationRegistry; diff --git a/backend/controlplane/src/integrations/model.rs b/backend/controlplane/src/integrations/model.rs new file mode 100644 index 0000000..db8e7c8 --- /dev/null +++ b/backend/controlplane/src/integrations/model.rs @@ -0,0 +1,201 @@ +//! Enterprise integration domain model. +//! +//! ClawForge governs *connections* to enterprise and government systems. The +//! control plane never stores secrets — only a [`CredentialRef`] pointing at +//! where the secret actually lives (a vault, an env var, an SSO provider). + +use chrono::Utc; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::constants::{LifecycleStatus, RiskLevel}; + +/// The category of an enterprise/government integration. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum IntegrationKind { + Oracle, + SqlServer, + Postgres, + MongoDb, + SharePoint, + ServiceNow, + ArcGis, + ActiveDirectory, + Sso, + ApiGateway, + Email, + Webhook, +} + +impl IntegrationKind { + /// Baseline risk for this category before considering granted permissions. + /// Identity stores and primary databases of record are the most sensitive. + pub fn default_risk(&self) -> RiskLevel { + use IntegrationKind::*; + match self { + ActiveDirectory | Sso => RiskLevel::Critical, + Oracle | SqlServer | Postgres | MongoDb => RiskLevel::High, + ServiceNow | SharePoint | ApiGateway => RiskLevel::Medium, + ArcGis | Email | Webhook => RiskLevel::Medium, + } + } +} + +/// Classify an integration's effective risk: the higher of the category +/// baseline and a floor implied by any elevated (write/delete/admin) permission. +pub fn classify_risk(kind: IntegrationKind, permissions: &[IntegrationPermission]) -> RiskLevel { + let base = kind.default_risk(); + let has_elevated = permissions.iter().any(|p| p.is_elevated()); + if has_elevated && base.weight() < RiskLevel::High.weight() { + RiskLevel::High + } else { + base + } +} + +/// Where an integration's credentials are kept. This is a *reference*, never +/// the secret material itself. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CredentialRef { + /// Backing secret store: `vault` | `env` | `sso` | `keychain` | `none`. + pub store: String, + /// Lookup key within that store (e.g. a vault path or env var name). + pub key: String, + /// Human-friendly note about the credential. + #[serde(default)] + pub description: String, +} + +impl CredentialRef { + /// A reference into a secrets vault. + pub fn vault(path: impl Into) -> Self { + CredentialRef { store: "vault".into(), key: path.into(), description: String::new() } + } + + /// A reference to an environment variable. + pub fn env(var: impl Into) -> Self { + CredentialRef { store: "env".into(), key: var.into(), description: String::new() } + } + + /// A placeholder for integrations that need no stored credential. + pub fn none() -> Self { + CredentialRef { store: "none".into(), key: String::new(), description: String::new() } + } + + /// Whether this reference actually points at a secret store. + pub fn is_present(&self) -> bool { + self.store != "none" && !self.key.is_empty() + } +} + +/// An operation an integration is permitted to perform. Granting `Write`, +/// `Delete`, or `Admin` is what elevates an integration's risk. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum IntegrationPermission { + /// Establish a connection / authenticate. + Connect, + /// Read data. + Read, + /// Write or update data. + Write, + /// Delete data. + Delete, + /// Administrative / privileged operations. + Admin, +} + +impl IntegrationPermission { + /// Whether this permission grants mutating or privileged access. + pub fn is_elevated(&self) -> bool { + matches!( + self, + IntegrationPermission::Write | IntegrationPermission::Delete | IntegrationPermission::Admin + ) + } +} + +/// An append-only audit entry for an integration's lifecycle. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IntegrationAuditEvent { + pub id: String, + pub integration_id: String, + /// `registered` | `approved` | `blocked`. + pub action: String, + pub at: i64, +} + +/// A registered enterprise/government integration and its governance metadata. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IntegrationProvider { + /// Stable UUID. + pub id: String, + pub name: String, + pub kind: IntegrationKind, + pub description: String, + /// Accountable owner. + pub owner: String, + /// Owning department. + pub department: String, + /// Connection endpoint (host, URL, or service identifier). + pub endpoint: String, + /// Reference to where credentials live (never the secret itself). + pub credential: CredentialRef, + /// Operations this integration is permitted to perform. + pub permissions: Vec, + /// Assessed risk level. + pub risk_level: RiskLevel, + /// Governance status (`pending_approval` / `active` / `blocked` / …). + pub status: LifecycleStatus, + pub created_at: i64, + pub updated_at: i64, +} + +/// Input used to register a new integration. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NewIntegration { + pub name: String, + pub kind: IntegrationKind, + pub description: String, + pub owner: String, + pub department: String, + pub endpoint: String, + pub credential: CredentialRef, + #[serde(default)] + pub permissions: Vec, + /// Explicit risk level; if omitted, callers can derive one from the kind. + pub risk_level: RiskLevel, +} + +impl IntegrationProvider { + /// Materialise a fresh integration record; starts in `PendingApproval`. + pub fn from_new(input: NewIntegration) -> Self { + let now = Utc::now().timestamp(); + IntegrationProvider { + id: Uuid::new_v4().to_string(), + name: input.name, + kind: input.kind, + description: input.description, + owner: input.owner, + department: input.department, + endpoint: input.endpoint, + credential: input.credential, + permissions: input.permissions, + risk_level: input.risk_level, + status: LifecycleStatus::PendingApproval, + created_at: now, + updated_at: now, + } + } + + /// Whether this integration may currently be used. + pub fn is_usable(&self) -> bool { + self.status.is_operational() + } + + /// Whether any granted permission is elevated (write/delete/admin). + pub fn has_elevated_permission(&self) -> bool { + self.permissions.iter().any(|p| p.is_elevated()) + } +} diff --git a/backend/controlplane/src/integrations/placeholders.rs b/backend/controlplane/src/integrations/placeholders.rs new file mode 100644 index 0000000..97859ac --- /dev/null +++ b/backend/controlplane/src/integrations/placeholders.rs @@ -0,0 +1,94 @@ +//! Placeholder constructors for common integration categories. +//! +//! These are convenience builders that produce a [`NewIntegration`] with +//! sensible defaults for each category. They are *connection blueprints*, not +//! live clients — the actual wire protocol is implemented by the runtime; the +//! control plane only governs the connection. + +use crate::constants::RiskLevel; + +use super::model::{CredentialRef, IntegrationKind, IntegrationPermission, NewIntegration}; + +/// Placeholder for a database integration (Oracle, SQL Server, Postgres, +/// MongoDB, …). Defaults to read-only `Connect`+`Read` at high risk. +pub fn database( + name: &str, + owner: &str, + department: &str, + kind: IntegrationKind, + endpoint: &str, + credential: CredentialRef, +) -> NewIntegration { + NewIntegration { + name: name.into(), + kind, + description: "Database integration".into(), + owner: owner.into(), + department: department.into(), + endpoint: endpoint.into(), + credential, + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Read], + risk_level: RiskLevel::High, + } +} + +/// Placeholder for a GIS integration (ArcGIS and similar spatial services). +pub fn gis(name: &str, owner: &str, department: &str, endpoint: &str, credential: CredentialRef) -> NewIntegration { + NewIntegration { + name: name.into(), + kind: IntegrationKind::ArcGis, + description: "GIS / spatial data integration".into(), + owner: owner.into(), + department: department.into(), + endpoint: endpoint.into(), + credential, + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Read], + risk_level: RiskLevel::Medium, + } +} + +/// Placeholder for an SSO / identity-provider integration (OIDC/SAML, AD). +/// Identity integrations are high risk by default. +pub fn sso(name: &str, owner: &str, department: &str, endpoint: &str, credential: CredentialRef) -> NewIntegration { + NewIntegration { + name: name.into(), + kind: IntegrationKind::Sso, + description: "Single sign-on / identity provider integration".into(), + owner: owner.into(), + department: department.into(), + endpoint: endpoint.into(), + credential, + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Read], + risk_level: RiskLevel::High, + } +} + +/// Placeholder for an email-sending integration (SMTP / provider API). +pub fn email(name: &str, owner: &str, department: &str, endpoint: &str, credential: CredentialRef) -> NewIntegration { + NewIntegration { + name: name.into(), + kind: IntegrationKind::Email, + description: "Outbound email integration".into(), + owner: owner.into(), + department: department.into(), + endpoint: endpoint.into(), + credential, + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Write], + risk_level: RiskLevel::Medium, + } +} + +/// Placeholder for an outbound webhook integration. +pub fn webhook(name: &str, owner: &str, department: &str, url: &str) -> NewIntegration { + NewIntegration { + name: name.into(), + kind: IntegrationKind::Webhook, + description: "Outbound webhook integration".into(), + owner: owner.into(), + department: department.into(), + endpoint: url.into(), + credential: CredentialRef::none(), + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Write], + risk_level: RiskLevel::Medium, + } +} diff --git a/backend/controlplane/src/integrations/store.rs b/backend/controlplane/src/integrations/store.rs new file mode 100644 index 0000000..8970baa --- /dev/null +++ b/backend/controlplane/src/integrations/store.rs @@ -0,0 +1,299 @@ +//! SQLite-backed integration registry. +//! +//! Tracks registered enterprise/government integrations, their governance +//! status, and the *reference* to their credentials (never the secret itself). + +use std::sync::Mutex; + +use chrono::Utc; +use rusqlite::{params, Connection}; +use uuid::Uuid; + +use crate::constants::LifecycleStatus; +use crate::error::{ControlPlaneError, Result}; + +use super::model::{classify_risk, IntegrationAuditEvent, IntegrationProvider, NewIntegration}; + +/// Persistent registry of enterprise integrations. +pub struct IntegrationRegistry { + pub(crate) conn: Mutex, +} + +const SCHEMA: &str = " + CREATE TABLE IF NOT EXISTS integrations ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + kind TEXT NOT NULL, + description TEXT NOT NULL, + owner TEXT NOT NULL, + department TEXT NOT NULL, + endpoint TEXT NOT NULL, + credential TEXT NOT NULL, + permissions TEXT NOT NULL, + risk_level TEXT NOT NULL, + status TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_integ_kind ON integrations(kind); + CREATE INDEX IF NOT EXISTS idx_integ_status ON integrations(status); + + CREATE TABLE IF NOT EXISTS integration_audit ( + id TEXT PRIMARY KEY, + integration_id TEXT NOT NULL, + action TEXT NOT NULL, + at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_integ_audit ON integration_audit(integration_id); +"; + +const COLUMNS: &str = "id, name, kind, description, owner, department, endpoint, credential, \ + permissions, risk_level, status, created_at, updated_at"; + +fn row_to_integration(row: &rusqlite::Row) -> rusqlite::Result { + Ok(IntegrationProvider { + id: row.get(0)?, + name: row.get(1)?, + kind: de(&row.get::<_, String>(2)?, 2)?, + description: row.get(3)?, + owner: row.get(4)?, + department: row.get(5)?, + endpoint: row.get(6)?, + credential: de(&row.get::<_, String>(7)?, 7)?, + permissions: de(&row.get::<_, String>(8)?, 8)?, + risk_level: de(&row.get::<_, String>(9)?, 9)?, + status: de(&row.get::<_, String>(10)?, 10)?, + created_at: row.get(11)?, + updated_at: row.get(12)?, + }) +} + +fn de(s: &str, col: usize) -> rusqlite::Result { + serde_json::from_str(s) + .map_err(|e| rusqlite::Error::FromSqlConversionFailure(col, rusqlite::types::Type::Text, Box::new(e))) +} + +impl IntegrationRegistry { + /// Open (creating if needed) a registry backed by a file. + pub fn open(path: &str) -> Result { + let conn = Connection::open(path)?; + conn.execute_batch(&format!("PRAGMA journal_mode=WAL;{SCHEMA}"))?; + Ok(Self { conn: Mutex::new(conn) }) + } + + /// Open an ephemeral in-memory registry (used by tests). + pub fn in_memory() -> Result { + let conn = Connection::open_in_memory()?; + conn.execute_batch(SCHEMA)?; + Ok(Self { conn: Mutex::new(conn) }) + } + + /// Register a new integration; it starts in `PendingApproval`. + pub fn register(&self, input: NewIntegration) -> Result { + if input.name.trim().is_empty() { + return Err(ControlPlaneError::validation("integration name must not be empty")); + } + let mut integration = IntegrationProvider::from_new(input); + // Escalate risk to at least the classified baseline for the kind and + // its granted permissions; never silently downgrade an explicit level. + let classified = classify_risk(integration.kind, &integration.permissions); + if classified.weight() > integration.risk_level.weight() { + integration.risk_level = classified; + } + self.upsert(&integration)?; + self.record_event(&integration.id, "registered")?; + cp_info!("integration.register", id = %integration.id, name = %integration.name); + Ok(integration) + } + + /// Append an audit event for an integration. + fn record_event(&self, integration_id: &str, action: &str) -> Result<()> { + let conn = self.conn.lock().expect("integration mutex poisoned"); + conn.execute( + "INSERT INTO integration_audit (id, integration_id, action, at) VALUES (?1,?2,?3,?4)", + params![Uuid::new_v4().to_string(), integration_id, action, Utc::now().timestamp()], + )?; + Ok(()) + } + + /// Full audit trail for an integration, oldest first. + pub fn audit_log(&self, integration_id: &str) -> Result> { + let conn = self.conn.lock().expect("integration mutex poisoned"); + let mut stmt = conn.prepare( + "SELECT id, integration_id, action, at FROM integration_audit + WHERE integration_id = ?1 ORDER BY at ASC", + )?; + let rows = stmt.query_map(params![integration_id], |row| { + Ok(IntegrationAuditEvent { + id: row.get(0)?, + integration_id: row.get(1)?, + action: row.get(2)?, + at: row.get(3)?, + }) + })?; + let mut out = Vec::new(); + for r in rows { + out.push(r?); + } + Ok(out) + } + + /// List all integrations, newest first. + pub fn list(&self) -> Result> { + let conn = self.conn.lock().expect("integration mutex poisoned"); + let mut stmt = conn.prepare(&format!("SELECT {COLUMNS} FROM integrations ORDER BY created_at DESC"))?; + let rows = stmt.query_map([], row_to_integration)?; + let mut out = Vec::new(); + for r in rows { + out.push(r?); + } + Ok(out) + } + + /// Fetch an integration by id. + pub fn get(&self, id: &str) -> Result { + let conn = self.conn.lock().expect("integration mutex poisoned"); + conn.query_row( + &format!("SELECT {COLUMNS} FROM integrations WHERE id = ?1"), + params![id], + row_to_integration, + ) + .map_err(|e| match e { + rusqlite::Error::QueryReturnedNoRows => ControlPlaneError::not_found("integration", id), + other => other.into(), + }) + } + + /// Approve an integration, making it usable (sets status `Active`). + pub fn approve(&self, id: &str) -> Result { + self.set_status(id, LifecycleStatus::Active) + } + + /// Block an integration, taking it out of service (sets status `Blocked`). + pub fn block(&self, id: &str) -> Result { + self.set_status(id, LifecycleStatus::Blocked) + } + + fn set_status(&self, id: &str, status: LifecycleStatus) -> Result { + let mut integration = self.get(id)?; + integration.status = status; + integration.updated_at = Utc::now().timestamp(); + self.upsert(&integration)?; + let action = match status { + LifecycleStatus::Active => "approved", + LifecycleStatus::Blocked => "blocked", + _ => "status_changed", + }; + self.record_event(id, action)?; + cp_info!("integration.status", id = %id, status = ?status); + Ok(integration) + } + + /// Total number of registered integrations. + pub fn count(&self) -> Result { + let conn = self.conn.lock().expect("integration mutex poisoned"); + let n: i64 = conn.query_row("SELECT COUNT(*) FROM integrations", [], |r| r.get(0))?; + Ok(n as u64) + } + + pub(crate) fn upsert(&self, i: &IntegrationProvider) -> Result<()> { + let conn = self.conn.lock().expect("integration mutex poisoned"); + conn.execute( + "INSERT OR REPLACE INTO integrations ( + id, name, kind, description, owner, department, endpoint, credential, + permissions, risk_level, status, created_at, updated_at + ) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13)", + params![ + i.id, + i.name, + serde_json::to_string(&i.kind)?, + i.description, + i.owner, + i.department, + i.endpoint, + serde_json::to_string(&i.credential)?, + serde_json::to_string(&i.permissions)?, + serde_json::to_string(&i.risk_level)?, + serde_json::to_string(&i.status)?, + i.created_at, + i.updated_at, + ], + )?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::constants::RiskLevel; + use crate::integrations::model::{CredentialRef, IntegrationKind, IntegrationPermission}; + + pub(super) fn input() -> NewIntegration { + NewIntegration { + name: "Resident DB".into(), + kind: IntegrationKind::Postgres, + description: "Resident records database".into(), + owner: "data-platform".into(), + department: "IT".into(), + endpoint: "postgres://db.internal:5432/residents".into(), + credential: CredentialRef::vault("kv/integrations/resident-db"), + permissions: vec![IntegrationPermission::Connect, IntegrationPermission::Read], + risk_level: RiskLevel::High, + } + } + + #[test] + fn register_list_get_lifecycle() { + let reg = IntegrationRegistry::in_memory().unwrap(); + let i = reg.register(input()).unwrap(); + assert_eq!(i.status, LifecycleStatus::PendingApproval); + assert!(!i.is_usable()); + assert_eq!(reg.list().unwrap().len(), 1); + assert_eq!(reg.get(&i.id).unwrap().name, "Resident DB"); + assert!(reg.approve(&i.id).unwrap().is_usable()); + assert!(!reg.block(&i.id).unwrap().is_usable()); + } + + #[test] + fn register_rejects_empty_name() { + let reg = IntegrationRegistry::in_memory().unwrap(); + let mut bad = input(); + bad.name = " ".into(); + assert!(reg.register(bad).is_err()); + } + + #[test] + fn risk_escalates_for_elevated_permissions() { + use crate::integrations::placeholders; + let reg = IntegrationRegistry::in_memory().unwrap(); + // A webhook is Medium by default, but Write permission floors it at High. + let i = reg + .register(placeholders::webhook("alerts", "ops", "IT", "https://hooks.internal/x")) + .unwrap(); + assert_eq!(i.risk_level, RiskLevel::High); + assert!(i.has_elevated_permission()); + } + + #[test] + fn sso_placeholder_is_critical() { + use crate::integrations::placeholders; + let reg = IntegrationRegistry::in_memory().unwrap(); + let i = reg + .register(placeholders::sso("corp-sso", "iam", "IT", "https://idp/realm", CredentialRef::vault("kv/sso"))) + .unwrap(); + assert_eq!(i.risk_level, RiskLevel::Critical); + assert!(i.credential.is_present()); + } + + #[test] + fn audit_log_tracks_lifecycle() { + let reg = IntegrationRegistry::in_memory().unwrap(); + let i = reg.register(input()).unwrap(); + reg.approve(&i.id).unwrap(); + reg.block(&i.id).unwrap(); + let log = reg.audit_log(&i.id).unwrap(); + let actions: Vec<&str> = log.iter().map(|e| e.action.as_str()).collect(); + assert_eq!(actions, vec!["registered", "approved", "blocked"]); + } +} diff --git a/backend/controlplane/src/lib.rs b/backend/controlplane/src/lib.rs index 40b974a..fd6b0dc 100644 --- a/backend/controlplane/src/lib.rs +++ b/backend/controlplane/src/lib.rs @@ -19,6 +19,7 @@ pub mod constants; pub mod error; pub mod gateway; pub mod governance; +pub mod integrations; pub mod marketplace; pub mod mcp; pub mod observability; diff --git a/docs/enterprise-integrations.md b/docs/enterprise-integrations.md new file mode 100644 index 0000000..6eb8200 --- /dev/null +++ b/docs/enterprise-integrations.md @@ -0,0 +1,68 @@ +# Enterprise Integrations + +ClawForge governs *connections* to enterprise and government systems. The +integration registry (`clawforge_controlplane::integrations`) tracks every +connector, who owns it, what it is allowed to do, where its credentials live, +and whether it is approved — **without ever storing a secret**. + +## Categories (`IntegrationKind`) + +Oracle, SQL Server, PostgreSQL, MongoDB, SharePoint, ServiceNow, ArcGIS, +Active Directory, SSO, API Gateway, Email, Webhook. + +## Credentials are referenced, never stored + +`CredentialRef` records *where* a secret lives (`vault` / `env` / `sso` / +`keychain` / `none`) and the lookup `key` — never the secret material itself. +This keeps the control plane out of scope for secret storage while still giving +governance a complete picture of what an integration can reach. + +```rust +CredentialRef::vault("kv/integrations/resident-db"); +CredentialRef::env("SMTP_PASSWORD"); +CredentialRef::none(); // e.g. an unauthenticated webhook +``` + +## Permissions & risk + +`IntegrationPermission` is `connect` / `read` / `write` / `delete` / `admin`. +Granting `write`, `delete`, or `admin` is *elevated*. `classify_risk(kind, +permissions)` computes effective risk: the higher of the category baseline +(identity stores and SSO are `critical`; primary databases are `high`) and a +`high` floor implied by any elevated permission. Registration escalates an +integration's risk to this baseline — it never silently downgrades an explicit +level. + +## API + +```rust +use clawforge_controlplane::integrations::{IntegrationRegistry, placeholders, CredentialRef}; +use clawforge_controlplane::integrations::model::IntegrationKind; + +let reg = IntegrationRegistry::open("clawforge-controlplane.db")?; + +// Placeholder builders for common categories: +let db = placeholders::database("Resident DB", "data", "IT", IntegrationKind::Postgres, + "postgres://db/residents", CredentialRef::vault("kv/db")); +let sso = placeholders::sso("Corp SSO", "iam", "IT", "https://idp/realm", CredentialRef::vault("kv/sso")); +let mail = placeholders::email("Notifier", "ops", "IT", "smtp://mail:587", CredentialRef::env("SMTP_PASSWORD")); +let gis = placeholders::gis("City GIS", "gis", "Planning", "https://gis/arcgis", CredentialRef::none()); +let hook = placeholders::webhook("Alerts", "ops", "IT", "https://hooks/x"); + +let integration = reg.register(db)?; // starts pending_approval +reg.approve(&integration.id)?; // make usable +// reg.block(&integration.id)? to take it out of service + +let trail = reg.audit_log(&integration.id)?; // registered → approved → blocked +``` + +## Audit + +Every register / approve / block appends an `IntegrationAuditEvent`, giving a +complete, ordered trail per integration for compliance evidence. + +## Status + +These are governance and connection *blueprints*. The actual wire protocol for +each system is implemented by the runtime; the control plane's job is to decide +which integrations exist, classify their risk, and keep them auditable.