diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1a236c42b66..55279c46b25 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -88,7 +88,7 @@ jobs:
         run: python -m pip install psycopg2-binary xmltodict
       - name: Run smoketests
         # Note: clear_database and replication only work in private
-        run: python -m smoketests ${{ matrix.smoketest_args }} -x clear_database replication
+        run: python -m smoketests ${{ matrix.smoketest_args }} -x clear_database replication teams
       - name: Stop containers (Linux)
         if: always() && runner.os == 'Linux'
         run: docker compose down
diff --git a/Cargo.lock b/Cargo.lock
index 36f07182a6c..061c293cd0d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7144,6 +7144,7 @@ dependencies = [
  "names",
  "notify 7.0.0",
  "percent-encoding",
+ "pretty_assertions",
  "quick-xml 0.31.0",
  "regex",
  "reqwest 0.12.24",
@@ -7169,6 +7170,7 @@ dependencies = [
  "tar",
  "tempfile",
  "termcolor",
+ "termtree",
  "thiserror 1.0.69",
  "tikv-jemalloc-ctl",
  "tikv-jemallocator",
@@ -8513,6 +8515,12 @@ dependencies = [
  "windows-sys 0.60.2",
 ]
 
+[[package]]
+name = "termtree"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683"
+
 [[package]]
 name = "test-client"
 version = "1.7.0"
diff --git a/Cargo.toml b/Cargo.toml
index d1dc2344c4d..75921b4a1aa 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -270,6 +270,7 @@ tar = "0.4"
 tempdir = "0.3.7"
 tempfile = "3.20"
 termcolor = "1.2.0"
+termtree = "0.5.1"
 thin-vec = "0.2.13"
 thiserror = "1.0.37"
 tokio = { version = "1.37", features = ["full"] }
diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml
index 04399c00090..5f93a58a9d2 100644
--- a/crates/cli/Cargo.toml
+++ b/crates/cli/Cargo.toml
@@ -64,6 +64,7 @@ tabled.workspace = true
 tar.workspace = true
 tempfile.workspace = true
 termcolor.workspace = true
+termtree.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tokio-tungstenite.workspace = true
@@ -83,6 +84,9 @@ quick-xml.workspace = true
 names.workspace = true
 notify.workspace = true
 
+[dev-dependencies]
+pretty_assertions.workspace = true
+
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemallocator = { workspace = true }
 tikv-jemalloc-ctl = { workspace = true }
diff --git a/crates/cli/src/subcommands/delete.rs b/crates/cli/src/subcommands/delete.rs
index e0d5c756137..b05e5d205cd 100644
--- a/crates/cli/src/subcommands/delete.rs
+++ b/crates/cli/src/subcommands/delete.rs
@@ -1,7 +1,15 @@
+use std::io;
+
 use crate::common_args;
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, database_identity, get_auth_header};
+use crate::util::{add_auth_header_opt, database_identity, get_auth_header, y_or_n, AuthHeader};
 use clap::{Arg, ArgMatches};
+use http::StatusCode;
+use itertools::Itertools as _;
+use reqwest::Response;
+use spacetimedb_client_api_messages::http::{DatabaseDeleteConfirmationResponse, DatabaseTree, DatabaseTreeNode};
+use spacetimedb_lib::Hash;
+use tokio::io::AsyncWriteExt as _;
 
 pub fn cli() -> clap::Command {
     clap::Command::new("delete")
@@ -22,11 +30,143 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     let force = args.get_flag("force");
 
     let identity = database_identity(&config, database, server).await?;
-
-    let builder = reqwest::Client::new().delete(format!("{}/v1/database/{}", config.get_host_url(server)?, identity));
+    let host_url = config.get_host_url(server)?;
+    let request_path = format!("{host_url}/v1/database/{identity}");
     let auth_header = get_auth_header(&mut config, false, server, !force).await?;
-    let builder = add_auth_header_opt(builder, &auth_header);
-    builder.send().await?.error_for_status()?;
+    let client = reqwest::Client::new();
+
+    let response = send_request(&client, &request_path, &auth_header, None).await?;
+    match response.status() {
+        StatusCode::PRECONDITION_REQUIRED => {
+            let confirm = response.json::<DatabaseDeleteConfirmationResponse>().await?;
+            println!("WARNING: Deleting the database {identity} will also delete its children!");
+            if !force {
+                print_database_tree_info(&confirm.database_tree).await?;
+            }
+            if y_or_n(force, "Do you want to proceed deleting above databases?")? {
+                send_request(&client, &request_path, &auth_header, Some(confirm.confirmation_token))
+                    .await?
+                    .error_for_status()?;
+            } else {
+                println!("Aborting");
+            }
+
+            Ok(())
+        }
+        StatusCode::OK => Ok(()),
+        _ => response.error_for_status().map(drop).map_err(Into::into),
+    }
+}
+
+async fn send_request(
+    client: &reqwest::Client,
+    request_path: &str,
+    auth: &AuthHeader,
+    confirmation_token: Option<Hash>,
+) -> Result<Response, reqwest::Error> {
+    let mut builder = client.delete(request_path);
+    builder = add_auth_header_opt(builder, auth);
+    if let Some(token) = confirmation_token {
+        builder = builder.query(&[("token", token)]);
+    }
+    builder.send().await
+}
+
+async fn print_database_tree_info(tree: &DatabaseTree) -> io::Result<()> {
+    tokio::io::stdout()
+        .write_all(as_termtree(tree).to_string().as_bytes())
+        .await
+}
+
+fn as_termtree(tree: &DatabaseTree) -> termtree::Tree<String> {
+    let mut stack: Vec<(&DatabaseTree, bool)> = vec![];
+    stack.push((tree, false));
+
+    let mut built: Vec<termtree::Tree<String>> = <_>::default();
+
+    while let Some((node, visited)) = stack.pop() {
+        if visited {
+            let mut term_node = termtree::Tree::new(fmt_tree_node(&node.root));
+            term_node.leaves = built.drain(built.len() - node.children.len()..).collect();
+            term_node.leaves.reverse();
+            built.push(term_node);
+        } else {
+            stack.push((node, true));
+            stack.extend(node.children.iter().rev().map(|child| (child, false)));
+        }
+    }
+
+    built
+        .pop()
+        .expect("database tree contains a root and we pushed it last")
+}
+
+fn fmt_tree_node(node: &DatabaseTreeNode) -> String {
+    format!(
+        "{}{}",
+        node.database_identity,
+        if node.database_names.is_empty() {
+            <_>::default()
+        } else {
+            format!(": {}", node.database_names.iter().join(", "))
+        }
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use spacetimedb_client_api_messages::http::{DatabaseTree, DatabaseTreeNode};
+    use spacetimedb_lib::{sats::u256, Identity};
 
-    Ok(())
+    #[test]
+    fn render_termtree() {
+        let tree = DatabaseTree {
+            root: DatabaseTreeNode {
+                database_identity: Identity::ONE,
+                database_names: ["parent".into()].into(),
+            },
+            children: vec![
+                DatabaseTree {
+                    root: DatabaseTreeNode {
+                        database_identity: Identity::from_u256(u256::new(2)),
+                        database_names: ["child".into()].into(),
+                    },
+                    children: vec![
+                        DatabaseTree {
+                            root: DatabaseTreeNode {
+                                database_identity: Identity::from_u256(u256::new(3)),
+                                database_names: ["grandchild".into()].into(),
+                            },
+                            children: vec![],
+                        },
+                        DatabaseTree {
+                            root: DatabaseTreeNode {
+                                database_identity: Identity::from_u256(u256::new(5)),
+                                database_names: [].into(),
+                            },
+                            children: vec![],
+                        },
+                    ],
+                },
+                DatabaseTree {
+                    root: DatabaseTreeNode {
+                        database_identity: Identity::from_u256(u256::new(4)),
+                        database_names: ["sibling".into(), "bro".into()].into(),
+                    },
+                    children: vec![],
+                },
+            ],
+        };
+        pretty_assertions::assert_eq!(
+            "\
+0000000000000000000000000000000000000000000000000000000000000001: parent
+├── 0000000000000000000000000000000000000000000000000000000000000004: bro, sibling
+└── 0000000000000000000000000000000000000000000000000000000000000002: child
+    ├── 0000000000000000000000000000000000000000000000000000000000000005
+    └── 0000000000000000000000000000000000000000000000000000000000000003: grandchild
+",
+            &as_termtree(&tree).to_string()
+        );
+    }
 }
diff --git a/crates/cli/src/subcommands/publish.rs b/crates/cli/src/subcommands/publish.rs
index 731858afe0d..91433d0c250 100644
--- a/crates/cli/src/subcommands/publish.rs
+++ b/crates/cli/src/subcommands/publish.rs
@@ -1,14 +1,15 @@
+use anyhow::{ensure, Context};
 use clap::Arg;
 use clap::ArgAction::{Set, SetTrue};
 use clap::ArgMatches;
 use reqwest::{StatusCode, Url};
 use spacetimedb_client_api_messages::name::{is_identity, parse_database_name, PublishResult};
-use spacetimedb_client_api_messages::name::{PrePublishResult, PrettyPrintStyle, PublishOp};
+use spacetimedb_client_api_messages::name::{DatabaseNameError, PrePublishResult, PrettyPrintStyle, PublishOp};
 use std::path::PathBuf;
 use std::{env, fs};
 
 use crate::config::Config;
-use crate::util::{add_auth_header_opt, get_auth_header, unauth_error_context, AuthHeader, ResponseExt};
+use crate::util::{add_auth_header_opt, get_auth_header, AuthHeader, ResponseExt};
 use crate::util::{decode_identity, y_or_n};
 use crate::{build, common_args};
 
@@ -75,6 +76,17 @@ pub fn cli() -> clap::Command {
         .arg(
             common_args::anonymous()
         )
+        .arg(
+            Arg::new("parent")
+            .help("Domain or identity of a parent for this database")
+            .long("parent")
+            .long_help(
+"A valid domain or identity of an existing database that should be the parent of this database.
+
+If a parent is given, the new database inherits the team permissions from the parent.
+A parent can only be set when a database is created, not when it is updated."
+            )
+        )
         .arg(
             Arg::new("name|identity")
                 .help("A valid domain or identity for this database")
@@ -106,6 +118,7 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     let build_options = args.get_one::<String>("build_options").unwrap();
     let num_replicas = args.get_one::<u8>("num_replicas");
     let break_clients_flag = args.get_flag("break_clients");
+    let parent = args.get_one::<String>("parent");
 
     // If the user didn't specify an identity and we didn't specify an anonymous identity, then
     // we want to use the default identity
@@ -113,6 +126,9 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     //  easily create a new identity with an email
     let auth_header = get_auth_header(&mut config, anon_identity, server, !force).await?;
 
+    let (name_or_identity, parent) =
+        validate_name_and_parent(name_or_identity.map(String::as_str), parent.map(String::as_str))?;
+
     if !path_to_project.exists() {
         return Err(anyhow::anyhow!(
             "Project path does not exist: {}",
@@ -152,14 +168,11 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     );
 
     let client = reqwest::Client::new();
-    // If a domain or identity was provided, we should locally make sure it looks correct and
+    // If a name was given, ensure to percent-encode it.
+    // We also use PUT with a name or identity, and POST otherwise.
     let mut builder = if let Some(name_or_identity) = name_or_identity {
-        if !is_identity(name_or_identity) {
-            parse_database_name(name_or_identity)?;
-        }
         let encode_set = const { &percent_encoding::NON_ALPHANUMERIC.remove(b'_').remove(b'-') };
         let domain = percent_encoding::percent_encode(name_or_identity.as_bytes(), encode_set);
-
         let mut builder = client.put(format!("{database_host}/v1/database/{domain}"));
 
         if !clear_database {
@@ -174,7 +187,7 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
                 break_clients_flag,
             )
             .await?;
-        };
+        }
 
         builder
     } else {
@@ -204,6 +217,9 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
         eprintln!("WARNING: Use of unstable option `--num-replicas`.\n");
         builder = builder.query(&[("num_replicas", *n)]);
     }
+    if let Some(parent) = parent {
+        builder = builder.query(&[("parent", parent)]);
+    }
 
     println!("Publishing module...");
 
@@ -220,18 +236,6 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     }
 
     let res = builder.body(program_bytes).send().await?;
-    if res.status() == StatusCode::UNAUTHORIZED && !anon_identity {
-        // If we're not in the `anon_identity` case, then we have already forced the user to log in above (using `get_auth_header`), so this should be safe to unwrap.
-        let token = config.spacetimedb_token().unwrap();
-        let identity = decode_identity(token)?;
-        let err = res.text().await?;
-        return unauth_error_context(
-            Err(anyhow::anyhow!(err)),
-            &identity,
-            config.server_nick_or_host(server)?,
-        );
-    }
-
     let response: PublishResult = res.json_or_error().await?;
     match response {
         PublishResult::Success {
@@ -270,6 +274,47 @@ pub async fn exec(mut config: Config, args: &ArgMatches) -> Result<(), anyhow::E
     Ok(())
 }
 
+fn validate_name_or_identity(name_or_identity: &str) -> Result<(), DatabaseNameError> {
+    if is_identity(name_or_identity) {
+        Ok(())
+    } else {
+        parse_database_name(name_or_identity).map(drop)
+    }
+}
+
+fn invalid_parent_name(name: &str) -> String {
+    format!("invalid parent database name `{name}`")
+}
+
+fn validate_name_and_parent<'a>(
+    name: Option<&'a str>,
+    parent: Option<&'a str>,
+) -> anyhow::Result<(Option<&'a str>, Option<&'a str>)> {
+    if let Some(parent) = parent.as_ref() {
+        validate_name_or_identity(parent).with_context(|| invalid_parent_name(parent))?;
+    }
+
+    match name {
+        Some(name) => match name.split_once('/') {
+            Some((parent_alt, child)) => {
+                ensure!(
+                    parent.is_none() || parent.is_some_and(|parent| parent == parent_alt),
+                    "cannot specify both --parent and <parent>/<child>"
+                );
+                validate_name_or_identity(parent_alt).with_context(|| invalid_parent_name(parent_alt))?;
+                validate_name_or_identity(child)?;
+
+                Ok((Some(child), Some(parent_alt)))
+            }
+            None => {
+                validate_name_or_identity(name)?;
+                Ok((Some(name), parent))
+            }
+        },
+        None => Ok((None, parent)),
+    }
+}
+
 /// Determine the pretty print style based on the NO_COLOR environment variable.
 ///
 /// See: https://no-color.org
@@ -293,16 +338,7 @@ async fn apply_pre_publish_if_needed(
     auth_header: &AuthHeader,
     break_clients_flag: bool,
 ) -> Result<reqwest::RequestBuilder, anyhow::Error> {
-    if let Some(pre) = call_pre_publish(
-        client,
-        base_url,
-        &domain.to_string(),
-        host_type,
-        program_bytes,
-        auth_header,
-    )
-    .await?
-    {
+    if let Some(pre) = call_pre_publish(client, base_url, domain, host_type, program_bytes, auth_header).await? {
         println!("{}", pre.migrate_plan);
 
         if pre.break_clients
@@ -359,3 +395,67 @@ async fn call_pre_publish(
     let pre_publish_result: PrePublishResult = res.json_or_error().await?;
     Ok(Some(pre_publish_result))
 }
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_matches;
+    use spacetimedb_lib::Identity;
+
+    use super::*;
+
+    #[test]
+    fn validate_none_arguments_returns_none_values() {
+        assert_matches!(validate_name_and_parent(None, None), Ok((None, None)));
+        assert_matches!(validate_name_and_parent(Some("foo"), None), Ok((Some(_), None)));
+        assert_matches!(validate_name_and_parent(None, Some("foo")), Ok((None, Some(_))));
+    }
+
+    #[test]
+    fn validate_valid_arguments_returns_arguments() {
+        let name = "child";
+        let parent = "parent";
+        let result = (Some(name), Some(parent));
+        assert_matches!(
+            validate_name_and_parent(Some(name), Some(parent)),
+            Ok(val) if val == result
+        );
+    }
+
+    #[test]
+    fn validate_parent_and_path_name_returns_error_unless_parent_equal() {
+        assert_matches!(
+            validate_name_and_parent(Some("parent/child"), Some("parent")),
+            Ok((Some("child"), Some("parent")))
+        );
+        assert_matches!(validate_name_and_parent(Some("parent/child"), Some("cousin")), Err(_));
+    }
+
+    #[test]
+    fn validate_more_than_two_path_segments_are_an_error() {
+        assert_matches!(validate_name_and_parent(Some("proc/net/tcp"), None), Err(_));
+        assert_matches!(validate_name_and_parent(Some("proc//net"), None), Err(_));
+    }
+
+    #[test]
+    fn validate_trailing_slash_is_an_error() {
+        assert_matches!(validate_name_and_parent(Some("foo//"), None), Err(_));
+        assert_matches!(validate_name_and_parent(Some("foo/bar/"), None), Err(_));
+    }
+
+    #[test]
+    fn validate_parent_cant_have_slash() {
+        assert_matches!(validate_name_and_parent(Some("child"), Some("par/ent")), Err(_));
+        assert_matches!(validate_name_and_parent(Some("child"), Some("parent/")), Err(_));
+    }
+
+    #[test]
+    fn validate_name_or_parent_can_be_identities() {
+        let parent = Identity::ZERO.to_string();
+        let child = Identity::ONE.to_string();
+
+        assert_matches!(
+            validate_name_and_parent(Some(&child), Some(&parent)),
+            Ok(res) if res == (Some(&child), Some(&parent))
+        );
+    }
+}
diff --git a/crates/cli/src/util.rs b/crates/cli/src/util.rs
index 74b483c54e6..df398ee30b8 100644
--- a/crates/cli/src/util.rs
+++ b/crates/cli/src/util.rs
@@ -281,15 +281,6 @@ pub fn y_or_n(force: bool, prompt: &str) -> anyhow::Result<bool> {
     Ok(input == "y" || input == "yes")
 }
 
-pub fn unauth_error_context<T>(res: anyhow::Result<T>, identity: &str, server: &str) -> anyhow::Result<T> {
-    res.with_context(|| {
-        format!(
-            "Identity {identity} is not valid for server {server}.
-Please log back in with `spacetime logout` and then `spacetime login`."
-        )
-    })
-}
-
 pub fn decode_identity(token: &String) -> anyhow::Result<String> {
     // Here, we manually extract and decode the claims from the json web token.
     // We do this without using the `jsonwebtoken` crate because it doesn't seem to have a way to skip signature verification.
diff --git a/crates/client-api-messages/src/http.rs b/crates/client-api-messages/src/http.rs
index fe966bf5dfc..02cc75968e6 100644
--- a/crates/client-api-messages/src/http.rs
+++ b/crates/client-api-messages/src/http.rs
@@ -1,6 +1,9 @@
+use std::collections::BTreeSet;
+use std::iter;
+
 use serde::{Deserialize, Serialize};
 use spacetimedb_lib::metrics::ExecutionMetrics;
-use spacetimedb_lib::ProductType;
+use spacetimedb_lib::{Hash, Identity, ProductType};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SqlStmtResult<Row> {
@@ -27,3 +30,34 @@ impl SqlStmtStats {
         }
     }
 }
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct DatabaseTree {
+    pub root: DatabaseTreeNode,
+    pub children: Vec<DatabaseTree>,
+}
+
+impl DatabaseTree {
+    pub fn iter(&self) -> impl Iterator<Item = &DatabaseTreeNode> + '_ {
+        let mut stack = vec![self];
+        iter::from_fn(move || {
+            let node = stack.pop()?;
+            for child in node.children.iter().rev() {
+                stack.push(child);
+            }
+            Some(&node.root)
+        })
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct DatabaseTreeNode {
+    pub database_identity: Identity,
+    pub database_names: BTreeSet<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseDeleteConfirmationResponse {
+    pub database_tree: DatabaseTree,
+    pub confirmation_token: Hash,
+}
diff --git a/crates/client-api/src/lib.rs b/crates/client-api/src/lib.rs
index 44fbe5f07d0..0d0f6d7cb1b 100644
--- a/crates/client-api/src/lib.rs
+++ b/crates/client-api/src/lib.rs
@@ -1,6 +1,9 @@
+use std::fmt;
+use std::future::Future;
 use std::num::NonZeroU8;
 use std::sync::Arc;
 
+use anyhow::anyhow;
 use async_trait::async_trait;
 use axum::response::ErrorResponse;
 use bytes::Bytes;
@@ -17,6 +20,7 @@ use spacetimedb_client_api_messages::name::{DomainName, InsertDomainResult, Regi
 use spacetimedb_lib::{ProductTypeElement, ProductValue};
 use spacetimedb_paths::server::ModuleLogsDir;
 use spacetimedb_schema::auto_migrate::{MigrationPolicy, PrettyPrintStyle};
+use thiserror::Error;
 use tokio::sync::watch;
 
 pub mod auth;
@@ -413,6 +417,92 @@ impl<T: NodeDelegate + ?Sized> NodeDelegate for Arc<T> {
     }
 }
 
+#[derive(Debug, Error)]
+pub enum Unauthorized {
+    #[error(
+        "{} is not authorized to perform action{}: {}",
+        subject,
+        database.map(|ident| format!(" on database {ident}")).unwrap_or_default(),
+        action
+    )]
+    Unauthorized {
+        subject: Identity,
+        action: Action,
+        // `Option` for future, non-database-bound actions.
+        database: Option<Identity>,
+        #[source]
+        source: Option<anyhow::Error>,
+    },
+    #[error("authorization failed due to internal error")]
+    InternalError(#[from] anyhow::Error),
+}
+
+impl axum::response::IntoResponse for Unauthorized {
+    fn into_response(self) -> axum::response::Response {
+        let (status, e) = match self {
+            unauthorized @ Self::Unauthorized { .. } => (StatusCode::UNAUTHORIZED, anyhow!(unauthorized)),
+            Self::InternalError(e) => {
+                log::error!("internal error: {e:#}");
+                (StatusCode::INTERNAL_SERVER_ERROR, e)
+            }
+        };
+
+        (status, format!("{e:#}")).into_response()
+    }
+}
+
+/// Action to be authorized via [Authorization::authorize_action].
+#[derive(Debug)]
+pub enum Action {
+    CreateDatabase { parent: Option<Identity> },
+    UpdateDatabase,
+    ResetDatabase,
+    DeleteDatabase,
+    RenameDatabase,
+    ViewModuleLogs,
+}
+
+impl fmt::Display for Action {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::CreateDatabase { parent } => match parent {
+                Some(parent) => write!(f, "create database with parent {}", parent),
+                None => f.write_str("create database"),
+            },
+            Self::UpdateDatabase => f.write_str("update database"),
+            Self::ResetDatabase => f.write_str("reset database"),
+            Self::DeleteDatabase => f.write_str("delete database"),
+            Self::RenameDatabase => f.write_str("rename database"),
+            Self::ViewModuleLogs => f.write_str("view module logs"),
+        }
+    }
+}
+
+/// Trait to delegate authorization of "actions" performed through the
+/// client API to an external, edition-specific implementation.
+pub trait Authorization {
+    /// Authorize `subject` to perform [Action] `action` on `database`.
+    ///
+    /// Return `Ok(())` if permission is granted, `Err(Unauthorized)` if denied.
+    fn authorize_action(
+        &self,
+        subject: Identity,
+        database: Identity,
+        action: Action,
+    ) -> impl Future<Output = Result<(), Unauthorized>> + Send;
+}
+
+impl<T: Authorization> Authorization for Arc<T> {
+    fn authorize_action(
+        &self,
+        subject: Identity,
+        database: Identity,
+        action: Action,
+    ) -> impl Future<Output = Result<(), Unauthorized>> + Send {
+        (**self).authorize_action(subject, database, action)
+    }
+}
+
 pub fn log_and_500(e: impl std::fmt::Display) -> ErrorResponse {
     log::error!("internal error: {e:#}");
     (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into()
diff --git a/crates/client-api/src/routes/database.rs b/crates/client-api/src/routes/database.rs
index 1700ca62aa5..48818096f1f 100644
--- a/crates/client-api/src/routes/database.rs
+++ b/crates/client-api/src/routes/database.rs
@@ -10,7 +10,10 @@ use crate::auth::{
 };
 use crate::routes::subscribe::generate_random_connection_id;
 pub use crate::util::{ByteStringBody, NameOrIdentity};
-use crate::{log_and_500, ControlStateDelegate, DatabaseDef, DatabaseResetDef, Host, NodeDelegate};
+use crate::{
+    log_and_500, Action, Authorization, ControlStateDelegate, DatabaseDef, DatabaseResetDef, Host, NodeDelegate,
+    Unauthorized,
+};
 use axum::body::{Body, Bytes};
 use axum::extract::{Path, Query, State};
 use axum::response::{ErrorResponse, IntoResponse};
@@ -426,7 +429,7 @@ pub async fn logs<S>(
     Extension(auth): Extension<SpacetimeAuth>,
 ) -> axum::response::Result<impl IntoResponse>
 where
-    S: ControlStateDelegate + NodeDelegate,
+    S: ControlStateDelegate + NodeDelegate + Authorization,
 {
     // You should not be able to read the logs from a database that you do not own
     // so, unless you are the owner, this will fail.
@@ -436,17 +439,9 @@ where
         .await?
         .ok_or(NO_SUCH_DATABASE)?;
 
-    if database.owner_identity != auth.claims.identity {
-        return Err((
-            StatusCode::BAD_REQUEST,
-            format!(
-                "Identity does not own database, expected: {} got: {}",
-                database.owner_identity.to_hex(),
-                auth.claims.identity.to_hex()
-            ),
-        )
-            .into());
-    }
+    worker_ctx
+        .authorize_action(auth.claims.identity, database.database_identity, Action::ViewModuleLogs)
+        .await?;
 
     let replica = worker_ctx
         .get_leader_replica_by_database(database.id)
@@ -617,7 +612,7 @@ pub struct ResetDatabaseQueryParams {
     host_type: HostType,
 }
 
-pub async fn reset<S: NodeDelegate + ControlStateDelegate>(
+pub async fn reset<S: NodeDelegate + ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(ResetDatabaseParams { name_or_identity }): Path<ResetDatabaseParams>,
     Query(ResetDatabaseQueryParams {
@@ -632,13 +627,8 @@ pub async fn reset<S: NodeDelegate + ControlStateDelegate>(
         .await?
         .ok_or(NO_SUCH_DATABASE)?;
 
-    if database.database_identity != auth.claims.identity {
-        return Err((
-            StatusCode::UNAUTHORIZED,
-            "only the database owner can reset / clear a database",
-        )
-            .into());
-    }
+    ctx.authorize_action(auth.claims.identity, database.database_identity, Action::ResetDatabase)
+        .await?;
 
     let num_replicas = num_replicas.map(validate_replication_factor).transpose()?.flatten();
     ctx.reset_database(
@@ -679,11 +669,10 @@ pub struct PublishDatabaseQueryParams {
     policy: MigrationPolicy,
     #[serde(default)]
     host_type: HostType,
-    #[allow(unused)] // ignore for now
     parent: Option<NameOrIdentity>,
 }
 
-pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
+pub async fn publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(PublishDatabaseParams { name_or_identity }): Path<PublishDatabaseParams>,
     Query(PublishDatabaseQueryParams {
@@ -692,7 +681,7 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
         token,
         policy,
         host_type,
-        parent: _, // ignore for now
+        parent,
     }): Query<PublishDatabaseQueryParams>,
     Extension(auth): Extension<SpacetimeAuth>,
     program_bytes: Bytes,
@@ -724,6 +713,10 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
     }
 
     let (database_identity, db_name) = get_or_create_identity_and_name(&ctx, &auth, name_or_identity.as_ref()).await?;
+    let maybe_parent_database_identity = match parent.as_ref() {
+        None => None,
+        Some(parent) => parent.resolve(&ctx).await.map(Some)?,
+    };
 
     // Check that the replication factor looks somewhat sane.
     let num_replicas = num_replicas.map(validate_replication_factor).transpose()?.flatten();
@@ -736,14 +729,19 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
         // If not, check that the we caller is sufficiently authenticated.
         None => {
             allow_creation(&auth)?;
+            if let Some(parent) = maybe_parent_database_identity {
+                ctx.authorize_action(
+                    auth.claims.identity,
+                    database_identity,
+                    Action::CreateDatabase { parent: Some(parent) },
+                )
+                .await?;
+            }
         }
         // If yes, authorize via ctx.
         Some(database) => {
-            // NOTE: This is redundant at the moment, but will be replaced by
-            // a more interesting authorization check with the "teams" feature.
-            if database.owner_identity != auth.claims.identity {
-                return Err((StatusCode::UNAUTHORIZED, "a database can only be updated by its owner").into());
-            }
+            ctx.authorize_action(auth.claims.identity, database.database_identity, Action::UpdateDatabase)
+                .await?;
         }
     }
 
@@ -753,6 +751,16 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
     } else {
         PublishOp::Created
     };
+    // If a parent is given, resolve to an existing database.
+    let parent = if let Some(name_or_identity) = parent {
+        let identity = name_or_identity
+            .resolve(&ctx)
+            .await
+            .map_err(|_| bad_request(format!("Parent database {name_or_identity} not found").into()))?;
+        Some(identity)
+    } else {
+        None
+    };
 
     let schema_migration_policy = schema_migration_policy(policy, token)?;
     let maybe_updated = ctx
@@ -763,7 +771,7 @@ pub async fn publish<S: NodeDelegate + ControlStateDelegate>(
                 program_bytes,
                 num_replicas,
                 host_type,
-                parent: None,
+                parent,
             },
             schema_migration_policy,
         )
@@ -895,7 +903,7 @@ pub struct PrePublishQueryParams {
     host_type: HostType,
 }
 
-pub async fn pre_publish<S: NodeDelegate + ControlStateDelegate>(
+pub async fn pre_publish<S: NodeDelegate + ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(PrePublishParams { name_or_identity }): Path<PrePublishParams>,
     Query(PrePublishQueryParams { style, host_type }): Query<PrePublishQueryParams>,
@@ -954,44 +962,39 @@ pub async fn pre_publish<S: NodeDelegate + ControlStateDelegate>(
 
 /// Resolves the [`NameOrIdentity`] to a database identity and checks if the
 /// `auth` identity owns the database.
-async fn resolve_and_authenticate<S: ControlStateDelegate>(
+async fn resolve_and_authenticate<S: ControlStateDelegate + Authorization>(
     ctx: &S,
     name_or_identity: &NameOrIdentity,
     auth: &SpacetimeAuth,
 ) -> axum::response::Result<Identity> {
     let database_identity = name_or_identity.resolve(ctx).await?;
-
     let database = worker_ctx_find_database(ctx, &database_identity)
         .await?
         .ok_or(NO_SUCH_DATABASE)?;
 
-    if database.owner_identity != auth.claims.identity {
-        return Err((
-            StatusCode::UNAUTHORIZED,
-            format!(
-                "Identity does not own database, expected: {} got: {}",
-                database.owner_identity.to_hex(),
-                auth.claims.identity.to_hex()
-            ),
-        )
-            .into());
-    }
+    ctx.authorize_action(auth.claims.identity, database.database_identity, Action::UpdateDatabase)
+        .await?;
 
     Ok(database_identity)
 }
 
 #[derive(Deserialize)]
 pub struct DeleteDatabaseParams {
-    name_or_identity: NameOrIdentity,
+    pub name_or_identity: NameOrIdentity,
 }
 
-pub async fn delete_database<S: ControlStateDelegate>(
+pub async fn delete_database<S: ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(DeleteDatabaseParams { name_or_identity }): Path<DeleteDatabaseParams>,
     Extension(auth): Extension<SpacetimeAuth>,
 ) -> axum::response::Result<impl IntoResponse> {
     let database_identity = name_or_identity.resolve(&ctx).await?;
+    let Some(_database) = worker_ctx_find_database(&ctx, &database_identity).await? else {
+        return Ok(());
+    };
 
+    ctx.authorize_action(auth.claims.identity, database_identity, Action::DeleteDatabase)
+        .await?;
     ctx.delete_database(&auth.claims.identity, &database_identity)
         .await
         .map_err(log_and_500)?;
@@ -1034,7 +1037,7 @@ pub struct SetNamesParams {
     name_or_identity: NameOrIdentity,
 }
 
-pub async fn set_names<S: ControlStateDelegate>(
+pub async fn set_names<S: ControlStateDelegate + Authorization>(
     State(ctx): State<S>,
     Path(SetNamesParams { name_or_identity }): Path<SetNamesParams>,
     Extension(auth): Extension<SpacetimeAuth>,
@@ -1057,14 +1060,18 @@ pub async fn set_names<S: ControlStateDelegate>(
         ));
     };
 
-    if database.owner_identity != auth.claims.identity {
-        return Ok((
-            StatusCode::UNAUTHORIZED,
-            axum::Json(name::SetDomainsResult::NotYourDatabase {
-                database: database.database_identity,
-            }),
-        ));
-    }
+    ctx.authorize_action(auth.claims.identity, database.database_identity, Action::RenameDatabase)
+        .await
+        .map_err(|e| match e {
+            Unauthorized::Unauthorized { .. } => (
+                StatusCode::UNAUTHORIZED,
+                axum::Json(name::SetDomainsResult::NotYourDatabase {
+                    database: database.database_identity,
+                }),
+            )
+                .into(),
+            Unauthorized::InternalError(e) => log_and_500(e),
+        })?;
 
     for name in &validated_names {
         if ctx.lookup_identity(name.as_str()).unwrap().is_some() {
@@ -1161,7 +1168,7 @@ pub struct DatabaseRoutes<S> {
 
 impl<S> Default for DatabaseRoutes<S>
 where
-    S: NodeDelegate + ControlStateDelegate + HasWebSocketOptions + Clone + 'static,
+    S: NodeDelegate + ControlStateDelegate + HasWebSocketOptions + Authorization + Clone + 'static,
 {
     fn default() -> Self {
         use axum::routing::{delete, get, post, put};
@@ -1189,7 +1196,7 @@ where
 
 impl<S> DatabaseRoutes<S>
 where
-    S: NodeDelegate + ControlStateDelegate + Clone + 'static,
+    S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static,
 {
     pub fn into_router(self, ctx: S) -> axum::Router<S> {
         let db_router = axum::Router::<S>::new()
diff --git a/crates/client-api/src/routes/mod.rs b/crates/client-api/src/routes/mod.rs
index f0930eefb4c..4fe1de67f41 100644
--- a/crates/client-api/src/routes/mod.rs
+++ b/crates/client-api/src/routes/mod.rs
@@ -1,8 +1,7 @@
-use database::DatabaseRoutes;
 use http::header;
 use tower_http::cors;
 
-use crate::{ControlStateDelegate, NodeDelegate};
+use crate::{Authorization, ControlStateDelegate, NodeDelegate};
 
 pub mod database;
 pub mod energy;
@@ -13,6 +12,8 @@ pub mod metrics;
 pub mod prometheus;
 pub mod subscribe;
 
+use self::database::DatabaseRoutes;
+
 /// This API call is just designed to allow clients to determine whether or not they can
 /// establish a connection to SpacetimeDB. This API call doesn't actually do anything.
 pub async fn ping(_auth: crate::auth::SpacetimeAuthHeader) {}
@@ -20,7 +21,7 @@ pub async fn ping(_auth: crate::auth::SpacetimeAuthHeader) {}
 #[allow(clippy::let_and_return)]
 pub fn router<S>(ctx: &S, database_routes: DatabaseRoutes<S>, extra: axum::Router<S>) -> axum::Router<S>
 where
-    S: NodeDelegate + ControlStateDelegate + Clone + 'static,
+    S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static,
 {
     use axum::routing::get;
     let router = axum::Router::new()
diff --git a/crates/standalone/src/lib.rs b/crates/standalone/src/lib.rs
index a2af70beb4a..1c56e9be046 100644
--- a/crates/standalone/src/lib.rs
+++ b/crates/standalone/src/lib.rs
@@ -20,7 +20,7 @@ use spacetimedb::util::jobs::JobCores;
 use spacetimedb::worker_metrics::WORKER_METRICS;
 use spacetimedb_client_api::auth::{self, LOCALHOST};
 use spacetimedb_client_api::routes::subscribe::{HasWebSocketOptions, WebSocketOptions};
-use spacetimedb_client_api::{DatabaseResetDef, Host, NodeDelegate};
+use spacetimedb_client_api::{ControlStateReadAccess as _, DatabaseResetDef, Host, NodeDelegate};
 use spacetimedb_client_api_messages::name::{DomainName, InsertDomainResult, RegisterTldResult, SetDomainsResult, Tld};
 use spacetimedb_datastore::db_metrics::data_size::DATA_SIZE_METRICS;
 use spacetimedb_datastore::db_metrics::DB_METRICS;
@@ -454,6 +454,30 @@ impl spacetimedb_client_api::ControlStateWriteAccess for StandaloneEnv {
     }
 }
 
+impl spacetimedb_client_api::Authorization for StandaloneEnv {
+    async fn authorize_action(
+        &self,
+        subject: Identity,
+        database: Identity,
+        action: spacetimedb_client_api::Action,
+    ) -> Result<(), spacetimedb_client_api::Unauthorized> {
+        let database = self
+            .get_database_by_identity(&database)?
+            .with_context(|| format!("database {database} not found"))
+            .with_context(|| format!("Unable to authorize {subject} to perform {action:?})"))?;
+        if subject == database.owner_identity {
+            return Ok(());
+        }
+
+        Err(spacetimedb_client_api::Unauthorized::Unauthorized {
+            subject,
+            action,
+            database: database.database_identity.into(),
+            source: None,
+        })
+    }
+}
+
 impl StandaloneEnv {
     async fn insert_replica(&self, replica: Replica) -> Result<(), anyhow::Error> {
         let mut new_replica = replica.clone();
diff --git a/smoketests/__init__.py b/smoketests/__init__.py
index 71e8cf46576..592269855d1 100644
--- a/smoketests/__init__.py
+++ b/smoketests/__init__.py
@@ -382,7 +382,7 @@ def tearDown(self):
         if "database_identity" in self.__dict__:
             try:
                 # TODO: save the credentials in publish_module()
-                self.spacetime("delete", self.database_identity)
+                self.spacetime("delete", "--yes", self.database_identity)
             except Exception:
                 pass
 
@@ -391,7 +391,7 @@ def tearDownClass(cls):
        if hasattr(cls, "database_identity"):
            try:
                # TODO: save the credentials in publish_module()
-               cls.spacetime("delete", cls.database_identity)
+               cls.spacetime("delete", "--yes", cls.database_identity)
            except Exception:
                pass
 
diff --git a/smoketests/tests/domains.py b/smoketests/tests/domains.py
index 61fef422f2f..8a39be046d5 100644
--- a/smoketests/tests/domains.py
+++ b/smoketests/tests/domains.py
@@ -1,5 +1,4 @@
 from .. import Smoketest, random_string
-import unittest
 import json
 
 class Domains(Smoketest):
@@ -26,15 +25,15 @@ def test_set_name(self):
         with self.assertRaises(Exception):
             self.spacetime("logs", orig_name)
 
-    @unittest.expectedFailure
     def test_subdomain_behavior(self):
         """Test how we treat the / character in published names"""
 
         root_name = random_string()
         self.publish_module(root_name)
-        id_to_rename = self.database_identity
 
-        self.publish_module(f"{root_name}/test")
+        # TODO: This is valid in editions with the teams feature, but
+        # smoketests don't know the target's edition.
+        # self.publish_module(f"{root_name}/test")
 
         with self.assertRaises(Exception):
             self.publish_module(f"{root_name}//test")
diff --git a/smoketests/tests/permissions.py b/smoketests/tests/permissions.py
index 39ffca6e022..3b1068d0c36 100644
--- a/smoketests/tests/permissions.py
+++ b/smoketests/tests/permissions.py
@@ -81,7 +81,7 @@ class PrivateTablePermissions(Smoketest):
     MODULE_CODE = """
 use spacetimedb::{ReducerContext, Table};
 
-#[spacetimedb::table(name = secret)]
+#[spacetimedb::table(name = secret, private)]
 pub struct Secret {
     answer: u8,
 }
@@ -97,9 +97,9 @@ class PrivateTablePermissions(Smoketest):
 }
 
 #[spacetimedb::reducer]
-pub fn do_thing(ctx: &ReducerContext) {
+pub fn do_thing(ctx: &ReducerContext, thing: String) {
     ctx.db.secret().insert(Secret { answer: 20 });
-    ctx.db.common_knowledge().insert(CommonKnowledge { thing: "howdy".to_owned() });
+    ctx.db.common_knowledge().insert(CommonKnowledge { thing });
 }
 """
 
@@ -113,7 +113,7 @@ def test_private_table(self):
             " 42     ",
             ""
         ])
-        self.assertMultiLineEqual(out, answer)
+        self.assertMultiLineEqual(str(out), answer)
 
         self.reset_config()
         self.new_identity()
@@ -121,12 +121,33 @@ def test_private_table(self):
         with self.assertRaises(Exception):
             self.spacetime("sql", self.database_identity, "select * from secret")
 
+        # Subscribing to the private table failes.
         with self.assertRaises(Exception):
             self.subscribe("SELECT * FROM secret", n=0)
 
+        # Subscribing to the public table works.
+        sub = self.subscribe("SELECT * FROM common_knowledge", n = 1)
+        self.call("do_thing", "godmorgon")
+        self.assertEqual(sub(), [
+            {
+                'common_knowledge': {
+                    'deletes': [],
+                    'inserts': [{'thing': 'godmorgon'}]
+                }
+            }
+        ])
+
+        # Subscribing to both tables returns updates for the public one.
         sub = self.subscribe("SELECT * FROM *", n=1)
-        self.call("do_thing", anon=True)
-        self.assertEqual(sub(), [{'common_knowledge': {'deletes': [], 'inserts': [{'thing': 'howdy'}]}}])
+        self.call("do_thing", "howdy", anon=True)
+        self.assertEqual(sub(), [
+            {
+                'common_knowledge': {
+                    'deletes': [],
+                    'inserts': [{'thing': 'howdy'}]
+                }
+            }
+        ])
 
 
 class LifecycleReducers(Smoketest):
diff --git a/smoketests/tests/teams.py b/smoketests/tests/teams.py
new file mode 100644
index 00000000000..0eeea186865
--- /dev/null
+++ b/smoketests/tests/teams.py
@@ -0,0 +1,335 @@
+import json
+import toml
+
+from .. import Smoketest, parse_sql_result, random_string
+
+class CreateChildDatabase(Smoketest):
+    AUTOPUBLISH = False
+
+    def test_create_child_database(self):
+        """
+        Test that the owner can add a child database,
+        and that deleting the parent also deletes the child.
+        """
+
+        parent_name = random_string()
+        child_name = random_string()
+
+        self.publish_module(parent_name)
+        parent_identity = self.database_identity
+        self.publish_module(f"{parent_name}/{child_name}")
+        child_identity = self.database_identity
+
+        databases = self.query_controldb(parent_identity, child_identity)
+        self.assertEqual(2, len(databases))
+
+        self.spacetime("delete", "--yes", parent_name)
+
+        databases = self.query_controldb(parent_identity, child_identity)
+        self.assertEqual(0, len(databases))
+
+    def query_controldb(self, parent, child):
+        res = self.spacetime(
+            "sql",
+            "spacetime-control",
+            f"select * from database where database_identity = 0x{parent} or database_identity = 0x{child}"
+        )
+        return parse_sql_result(str(res))
+
+class PermissionsTest(Smoketest):
+    AUTOPUBLISH = False
+
+    def create_identity(self):
+        """
+        Obtain a fresh identity and token from the server.
+        Doesn't alter the config.toml for this test instance.
+        """
+        resp = self.api_call("POST", "/v1/identity")
+        return json.loads(resp)
+
+    def create_collaborators(self, database):
+        """
+        Create collaborators for the current database, one for each role.
+        """
+        collaborators = {}
+        roles = ["Owner", "Admin", "Developer", "Viewer"]
+        for role in roles:
+            identity_and_token = self.create_identity()
+            self.call_controldb_reducer(
+                "upsert_collaborator",
+                {"Name": database},
+                [f"0x{identity_and_token['identity']}"],
+                {role: {}}
+            )
+            collaborators[role] = identity_and_token
+        return collaborators
+
+
+    def call_controldb_reducer(self, reducer, *args):
+        """
+        Call a controldb reducer.
+        """
+        self.spacetime("call", "spacetime-control", reducer, *map(json.dumps, args))
+
+    def login_with(self, identity_and_token: dict):
+        self.spacetime("logout")
+        config = toml.load(self.config_path)
+        config['spacetimedb_token'] = identity_and_token['token']
+        with open(self.config_path, 'w') as f:
+            toml.dump(config, f)
+
+    def publish_as(self, role_and_token, module, code, clear = False):
+        print(f"publishing {module} as {role_and_token[0]}:")
+        print(f"{code}")
+        self.login_with(role_and_token[1])
+        self.write_module_code(code)
+        self.publish_module(module, clear = clear)
+        return self.database_identity
+
+    def sql_as(self, role_and_token, database, sql):
+        """
+        Log in as `token` and run an SQL statement against `database`
+        """
+        print(f"running sql as {role_and_token[0]}: {sql}")
+        self.login_with(role_and_token[1])
+        res = self.spacetime("sql", database, sql)
+        return parse_sql_result(str(res))
+
+    def subscribe_as(self, role_and_token, *queries, n):
+        """
+        Log in as `token` and subscribe to the current database using `queries`.
+        """
+        print(f"subscribe as {role_and_token[0]}: {queries}")
+        self.login_with(role_and_token[1])
+        return self.subscribe(*queries, n = n)
+
+
+class MutableSql(PermissionsTest):
+    MODULE_CODE = """
+#[spacetimedb::table(name = person, public)]
+struct Person {
+    name: String,
+}
+"""
+    def test_permissions_for_mutable_sql_transactions(self):
+        """
+        Tests that only owners and admins can perform mutable SQL transactions.
+        """
+
+        name = random_string()
+        self.publish_module(name)
+        team = self.create_collaborators(name)
+
+        for role, token in team.items():
+            self.login_with(token)
+            dml = f"insert into person (name) values ('bob-the-{role}')"
+            if role == "Owner" or role == "Admin":
+                self.spacetime("sql", name, dml)
+            else:
+                with self.assertRaises(Exception):
+                    self.spacetime("sql", name, dml)
+
+
+class PublishDatabase(PermissionsTest):
+    MODULE_CODE = """
+#[spacetimedb::table(name = person, public)]
+struct Person {
+    name: String,
+}
+"""
+
+    MODULE_CODE_OWNER = MODULE_CODE + """
+#[spacetimedb::table(name = owner)]
+struct Owner {
+    name: String,
+}
+"""
+
+    MODULE_CODE_ADMIN = MODULE_CODE_OWNER + """
+#[spacetimedb::table(name = admin)]
+struct Admin {
+    name: String,
+}
+"""
+
+    MODULE_CODE_DEVELOPER = MODULE_CODE_ADMIN + """
+#[spacetimedb::table(name = developer)]
+struct Developer {
+    name: String,
+}
+"""
+
+    MODULE_CODE_VIEWER = MODULE_CODE_DEVELOPER + """
+#[spacetimedb::table(name = viewer)]
+struct Viewer {
+    name: String,
+}
+"""
+
+    def test_permissions_publish(self):
+        """
+        Tests that only owner, admin and developer roles can publish a database.
+        """
+
+        parent = random_string()
+        self.publish_module(parent)
+
+        (owner, admin, developer, viewer) = self.create_collaborators(parent).items()
+        succeed_with = [
+            (owner, self.MODULE_CODE_OWNER),
+            (admin, self.MODULE_CODE_ADMIN),
+            (developer, self.MODULE_CODE_DEVELOPER)
+        ]
+
+        for role_and_token, code in succeed_with:
+            self.publish_as(role_and_token, parent, code)
+
+        with self.assertRaises(Exception):
+            self.publish_as(viewer, parent, self.MODULE_CODE_VIEWER)
+
+        # Create a child database.
+        child = random_string()
+        child_path = f"{parent}/{child}"
+
+        # Developer and viewer should not be able to create a child.
+        for role_and_token in [developer, viewer]:
+            with self.assertRaises(Exception):
+                self.publish_as(role_and_token, child_path, self.MODULE_CODE)
+        # But admin should succeed.
+        self.publish_as(admin, child_path, self.MODULE_CODE)
+
+        # Once created, only viewer should be denied updating.
+        for role_and_token, code in succeed_with:
+            self.publish_as(role_and_token, child_path, code)
+
+        with self.assertRaises(Exception):
+            self.publish_as(viewer, child_path, self.MODULE_CODE_VIEWER)
+
+
+class ClearDatabase(PermissionsTest):
+    def test_permissions_clear(self):
+        """
+        Tests that only owners and admins can clear a database.
+        """
+
+        parent = random_string()
+        self.publish_module(parent)
+        # First degree owner can clear.
+        self.publish_module(parent, clear = True)
+
+        (owner, admin, developer, viewer) = self.create_collaborators(parent).items()
+
+        # Owner and admin collaborators can clear.
+        for role_and_token in [owner, admin]:
+            self.publish_as(role_and_token, parent, self.MODULE_CODE, clear = True)
+
+        # Others can't.
+        for role_and_token in [developer, viewer]:
+            with self.assertRaises(Exception):
+                self.publish_as(role_and_token, parent, self.MODULE_CODE, clear = True)
+
+        # Same applies to child.
+        child = random_string()
+        child_path = f"{parent}/{child}"
+
+        self.publish_as(owner, child_path, self.MODULE_CODE)
+
+        for role_and_token in [owner, admin]:
+            self.publish_as(role_and_token, parent, self.MODULE_CODE, clear = True)
+
+        for role_and_token in [developer, viewer]:
+            with self.assertRaises(Exception):
+                self.publish_as(role_and_token, parent, self.MODULE_CODE, clear = True)
+
+
+class DeleteDatabase(PermissionsTest):
+    def delete_as(self, role_and_token, database):
+        print(f"delete {database} as {role_and_token[0]}")
+        self.login_with(role_and_token[1])
+        self.spacetime("delete", "--yes", database)
+
+    def test_permissions_delete(self):
+        """
+        Tests that only owners can delete databases.
+        """
+
+        parent = random_string()
+        self.publish_module(parent)
+        self.spacetime("delete", "--yes", parent)
+
+        self.publish_module(parent)
+
+        (owner, admin, developer, viewer) = self.create_collaborators(parent).items()
+
+        for role_and_token in [admin, developer, viewer]:
+            with self.assertRaises(Exception):
+                self.delete_as(role_and_token, parent)
+
+        child = random_string()
+        child_path = f"{parent}/{child}"
+
+        # If admin creates a child, they should also be able to delete it,
+        # because they are the owner of the child.
+        print("publish and delete as admin")
+        self.publish_as(admin, child_path, self.MODULE_CODE)
+        self.delete_as(admin, child)
+
+        # The owner role should be able to delete.
+        print("publish as admin, delete as owner")
+        self.publish_as(admin, child_path, self.MODULE_CODE)
+        self.delete_as(owner, child)
+
+        # Anyone else should be denied if not direct owner.
+        print("publish as owner, deny deletion by admin, developer, viewer")
+        self.publish_as(owner, child_path, self.MODULE_CODE)
+        for role_and_token in [admin, developer, viewer]:
+            with self.assertRaises(Exception):
+                self.delete_as(role_and_token, child)
+
+        print("delete child as owner")
+        self.delete_as(owner, child)
+
+        print("delete parent as owner")
+        self.delete_as(owner, parent)
+
+
+class PrivateTables(PermissionsTest):
+    def test_permissions_private_tables(self):
+        """
+        Test that all collaborators can read private tables.
+        """
+
+        parent = random_string()
+        self.publish_module(parent)
+
+        team = self.create_collaborators(parent)
+        owner = ("Owner", team['Owner'])
+
+        self.sql_as(owner, parent, "insert into person (name) values ('horsti')")
+
+        for role_and_token in team.items():
+            rows = self.sql_as(role_and_token, parent, "select * from person")
+            self.assertEqual(rows, [{ "name": '"horsti"' }])
+
+        for role_and_token in team.items():
+            sub = self.subscribe_as(role_and_token, "select * from person", n = 2)
+            self.sql_as(owner, parent, "insert into person (name) values ('hansmans')")
+            self.sql_as(owner, parent, "delete from person where name = 'hansmans'")
+            res = sub()
+            self.assertEqual(
+                res,
+                [
+                    {
+                        'person': {
+                            'deletes': [],
+                            'inserts': [{'name': 'hansmans'}]
+                        }
+                    },
+                    {
+                        'person': {
+                            'deletes': [{'name': 'hansmans'}],
+                            'inserts': []
+                        }
+                    }
+                ],
+            )