use sqlx::{PgPool, Row};
use crate::config::AdminConfig;
use crate::contract_sql::{CONTRACT_SQL, CONTRACT_SQL_SHA256};
use crate::error::{KernelError, Result};
const BACKEND_MIGRATIONS: &[&str] = &[
include_str!("../migrations/0001_kernel_internal.sql"),
include_str!("../migrations/0002_writer.sql"),
include_str!("../migrations/0003_blob.sql"),
include_str!("../migrations/0004_checkpoint.sql"),
];
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TargetState {
Empty,
Initialized { contract_sha256: String },
Foreign { objects: Vec<String> },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InitOutcome {
Initialized,
AlreadyInitialized,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RoleAttributes {
pub superuser: bool,
pub create_role: bool,
pub create_db: bool,
pub bypass_rls: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RuntimePrivileges {
pub attributes: RoleAttributes,
pub can_update_event: bool,
pub can_delete_event: bool,
pub can_update_receipt: bool,
pub can_delete_receipt: bool,
pub can_create_in_gwk: bool,
}
impl RoleAttributes {
pub fn violations(&self) -> Vec<&'static str> {
let mut out = Vec::new();
if self.superuser {
out.push("SUPERUSER");
}
if self.create_role {
out.push("CREATEROLE");
}
if self.create_db {
out.push("CREATEDB");
}
if self.bypass_rls {
out.push("BYPASSRLS");
}
out
}
}
impl RuntimePrivileges {
pub fn violations(&self) -> Vec<&'static str> {
let mut out = self.attributes.violations();
if self.can_update_event {
out.push("UPDATE on gwk.event");
}
if self.can_delete_event {
out.push("DELETE on gwk.event");
}
if self.can_update_receipt {
out.push("UPDATE on gwk.receipt");
}
if self.can_delete_receipt {
out.push("DELETE on gwk.receipt");
}
if self.can_create_in_gwk {
out.push("CREATE on schema gwk");
}
out
}
}
pub fn classify(
gwk_present: bool,
contract_sha256: Option<String>,
foreign_objects: Vec<String>,
) -> TargetState {
if !foreign_objects.is_empty() {
return TargetState::Foreign {
objects: foreign_objects,
};
}
match (gwk_present, contract_sha256) {
(false, None) => TargetState::Empty,
(true, Some(contract_sha256)) => TargetState::Initialized { contract_sha256 },
(true, None) => TargetState::Foreign {
objects: vec!["schema gwk (without a gwk_internal.schema_fingerprint row)".to_owned()],
},
(false, Some(_)) => TargetState::Foreign {
objects: vec!["gwk_internal.schema_fingerprint (without a gwk schema)".to_owned()],
},
}
}
pub async fn inspect(pool: &PgPool) -> Result<TargetState> {
let row = sqlx::query(
"SELECT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'gwk') AS gwk_present, \
to_regclass('gwk_internal.schema_fingerprint') IS NOT NULL AS fingerprint_table",
)
.fetch_one(pool)
.await?;
let gwk_present: bool = row.try_get("gwk_present")?;
let fingerprint_table: bool = row.try_get("fingerprint_table")?;
let contract_sha256: Option<String> = if fingerprint_table {
sqlx::query_scalar(
"SELECT contract_sha256 FROM gwk_internal.schema_fingerprint WHERE id = 1",
)
.fetch_optional(pool)
.await?
} else {
None
};
let foreign_objects: Vec<String> = sqlx::query_scalar(
"SELECT n.nspname || '.' || c.relname \
FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE c.relkind IN ('r', 'p', 'v', 'm', 'S', 'f') \
AND n.nspname NOT IN ('information_schema', 'gwk', 'gwk_internal') \
AND n.nspname NOT LIKE 'pg\\_%' \
ORDER BY 1 LIMIT 20",
)
.fetch_all(pool)
.await?;
Ok(classify(gwk_present, contract_sha256, foreign_objects))
}
pub async fn role_attributes<'e>(
executor: impl sqlx::PgExecutor<'e>,
role: &str,
) -> Result<Option<RoleAttributes>> {
let row = sqlx::query(
"SELECT rolsuper, rolcreaterole, rolcreatedb, rolbypassrls \
FROM pg_roles WHERE rolname = $1",
)
.bind(role)
.fetch_optional(executor)
.await?;
row.map(|row| {
Ok(RoleAttributes {
superuser: row.try_get("rolsuper")?,
create_role: row.try_get("rolcreaterole")?,
create_db: row.try_get("rolcreatedb")?,
bypass_rls: row.try_get("rolbypassrls")?,
})
})
.transpose()
}
pub async fn runtime_privileges<'e>(
executor: impl sqlx::PgExecutor<'e>,
) -> Result<RuntimePrivileges> {
let row = sqlx::query(
"SELECT rolsuper, rolcreaterole, rolcreatedb, rolbypassrls, \
has_table_privilege('gwk.event', 'UPDATE') AS upd_event, \
has_table_privilege('gwk.event', 'DELETE') AS del_event, \
has_table_privilege('gwk.receipt', 'UPDATE') AS upd_receipt, \
has_table_privilege('gwk.receipt', 'DELETE') AS del_receipt, \
has_schema_privilege('gwk', 'CREATE') AS create_gwk \
FROM pg_roles WHERE rolname = current_user",
)
.fetch_one(executor)
.await?;
Ok(RuntimePrivileges {
attributes: RoleAttributes {
superuser: row.try_get("rolsuper")?,
create_role: row.try_get("rolcreaterole")?,
create_db: row.try_get("rolcreatedb")?,
bypass_rls: row.try_get("rolbypassrls")?,
},
can_update_event: row.try_get("upd_event")?,
can_delete_event: row.try_get("del_event")?,
can_update_receipt: row.try_get("upd_receipt")?,
can_delete_receipt: row.try_get("del_receipt")?,
can_create_in_gwk: row.try_get("create_gwk")?,
})
}
pub fn backend_script(role: &str, contract_sha256: &str) -> String {
let migrations = BACKEND_MIGRATIONS.join("\n");
format!(
"{migrations}\n\
INSERT INTO gwk_internal.schema_fingerprint (id, contract_sha256) \
VALUES (1, '{contract_sha256}');\n\
GRANT USAGE ON SCHEMA gwk TO {role};\n\
GRANT USAGE ON SCHEMA gwk_internal TO {role};\n\
GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA gwk TO {role};\n\
REVOKE UPDATE ON gwk.event, gwk.receipt, gwk.ingested_record FROM {role};\n\
REVOKE INSERT, UPDATE ON gwk.transition FROM {role};\n\
GRANT SELECT ON gwk_internal.schema_fingerprint TO {role};\n\
GRANT SELECT, UPDATE ON gwk_internal.writer TO {role};\n\
GRANT SELECT, INSERT, UPDATE, DELETE ON \
gwk_internal.blob, gwk_internal.blob_pin, gwk_internal.blob_upload TO {role};\n\
GRANT SELECT, INSERT ON gwk_internal.checkpoint TO {role};\n"
)
}
pub async fn init(pool: &PgPool, admin: &AdminConfig) -> Result<InitOutcome> {
let role = admin.runtime_role();
let attributes = role_attributes(pool, role).await?.ok_or_else(|| {
KernelError::Privilege(format!(
"role {role:?} does not exist: initialization grants an already-created role and \
never creates one"
))
})?;
let violations = attributes.violations();
if !violations.is_empty() {
return Err(KernelError::Privilege(format!(
"role {role:?} holds {}: the kernel refuses to run as a role that can re-grant or \
re-DDL its own store",
violations.join(", ")
)));
}
match inspect(pool).await? {
TargetState::Initialized { contract_sha256 } if contract_sha256 == CONTRACT_SQL_SHA256 => {
return Ok(InitOutcome::AlreadyInitialized);
}
TargetState::Initialized { contract_sha256 } => {
return Err(KernelError::Schema(format!(
"this database carries contract {contract_sha256}, and this binary carries \
{CONTRACT_SQL_SHA256} — initialize a fresh database with the matching build"
)));
}
TargetState::Foreign { objects } => {
return Err(KernelError::Schema(format!(
"refusing to initialize a nonempty database this binary does not recognize; it \
already contains: {}. Create a fresh empty database and point \
GWK_ADMIN_DATABASE_URL at that",
objects.join(", ")
)));
}
TargetState::Empty => {}
}
sqlx::raw_sql(CONTRACT_SQL).execute(pool).await?;
sqlx::raw_sql(sqlx::AssertSqlSafe(backend_script(
role,
CONTRACT_SQL_SHA256,
)))
.execute(pool)
.await?;
Ok(InitOutcome::Initialized)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn an_empty_database_is_the_only_thing_init_will_write_to() {
assert_eq!(classify(false, None, vec![]), TargetState::Empty);
assert_eq!(
classify(true, Some("abc".to_owned()), vec![]),
TargetState::Initialized {
contract_sha256: "abc".to_owned()
}
);
}
#[test]
fn every_incoherent_or_occupied_target_is_refused() {
let foreign = classify(
true,
Some("abc".to_owned()),
vec!["public.users".to_owned()],
);
assert_eq!(
foreign,
TargetState::Foreign {
objects: vec!["public.users".to_owned()]
}
);
for state in [
classify(true, None, vec![]),
classify(false, Some("abc".to_owned()), vec![]),
] {
assert!(
matches!(state, TargetState::Foreign { .. }),
"expected a refusal, got {state:?}"
);
}
}
#[test]
fn a_role_that_outranks_the_kernel_names_every_reason() {
let clean = RoleAttributes {
superuser: false,
create_role: false,
create_db: false,
bypass_rls: false,
};
assert!(clean.violations().is_empty());
let all = RoleAttributes {
superuser: true,
create_role: true,
create_db: true,
bypass_rls: true,
};
assert_eq!(
all.violations(),
["SUPERUSER", "CREATEROLE", "CREATEDB", "BYPASSRLS"]
);
let mut privileges = RuntimePrivileges {
attributes: clean,
can_update_event: false,
can_delete_event: false,
can_update_receipt: false,
can_delete_receipt: false,
can_create_in_gwk: false,
};
assert!(privileges.violations().is_empty());
privileges.can_delete_event = true;
privileges.can_create_in_gwk = true;
assert_eq!(
privileges.violations(),
["DELETE on gwk.event", "CREATE on schema gwk"]
);
}
#[test]
fn the_grant_script_withholds_history_mutation_and_all_deletion() {
let script = backend_script("gwk_runtime", &"a".repeat(64));
assert!(script.contains("CREATE SCHEMA IF NOT EXISTS gwk_internal;"));
assert!(script.contains("VALUES (1, '"));
assert!(script.contains("GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA gwk"));
assert!(script.contains("REVOKE UPDATE ON gwk.event, gwk.receipt"));
assert!(script.contains("REVOKE INSERT, UPDATE ON gwk.transition"));
assert!(!script.contains("TRUNCATE"), "{script}");
let granted: Vec<&str> = script
.lines()
.filter(|line| line.starts_with("GRANT") && line.contains("DELETE"))
.collect();
assert_eq!(granted.len(), 1, "{script}");
for object in ["gwk_internal.blob", "gwk_internal.blob_pin"] {
assert!(granted[0].contains(object), "{}", granted[0]);
}
assert!(!granted[0].contains(" gwk."), "{}", granted[0]);
}
}