Compare commits
2 Commits
main
..
8d95b79db6
| Author | SHA1 | Date | |
|---|---|---|---|
| 8d95b79db6 | |||
| 65d71b6bfd |
+20
-33
@@ -7,47 +7,34 @@ platform:
|
||||
|
||||
steps:
|
||||
- name: build-release
|
||||
image: rust:1.93
|
||||
image: rust:1.91
|
||||
commands:
|
||||
- cargo build --verbose --workspace --release
|
||||
- name: test
|
||||
image: rust:1.93
|
||||
image: rust:1.91
|
||||
commands:
|
||||
- cargo test --verbose --workspace
|
||||
- name: lint
|
||||
image: rust:1.93
|
||||
image: rust:1.91
|
||||
commands:
|
||||
- rustup component add clippy
|
||||
- cargo clippy --all-targets --all-features
|
||||
- name: format
|
||||
image: rust:1.93
|
||||
commands:
|
||||
- rustup component add rustfmt
|
||||
- cargo fmt --all -- --check
|
||||
- name: build-image
|
||||
image: plugins/docker
|
||||
- name: push-binary-release
|
||||
image: plugins/gitea-release
|
||||
settings:
|
||||
repo: gitea.lardenois.cc/matos-ai/pg-instance-handler
|
||||
tags: latest
|
||||
dockerfile: Dockerfile
|
||||
dry_run: "true"
|
||||
base_url: https://gitea.lardenois.cc
|
||||
api_key:
|
||||
from_secret: gitea-package
|
||||
files:
|
||||
- target/release/pg-instance-handler
|
||||
title: "Release ${DRONE_TAG:-v${DRONE_BUILD_NUMBER}}"
|
||||
note: "Automated release from commit ${DRONE_COMMIT_SHA}"
|
||||
checksum:
|
||||
- sha256
|
||||
when:
|
||||
branch:
|
||||
exclude:
|
||||
- main
|
||||
- name: build-and-push-image
|
||||
image: plugins/docker
|
||||
settings:
|
||||
registry: gitea.lardenois.cc
|
||||
repo: gitea.lardenois.cc/matos-ai/pg-instance-handler
|
||||
tags:
|
||||
- latest
|
||||
dockerfile: Dockerfile
|
||||
username:
|
||||
from_secret: GITEA_USER
|
||||
password:
|
||||
from_secret: GITEA_PASSWORD
|
||||
when:
|
||||
branch:
|
||||
include:
|
||||
- main
|
||||
event: tag
|
||||
status: success
|
||||
depends_on:
|
||||
- build-release
|
||||
- test
|
||||
- lint
|
||||
|
||||
Generated
-45
@@ -23,17 +23,6 @@ version = "1.0.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78"
|
||||
|
||||
[[package]]
|
||||
name = "async-trait"
|
||||
version = "0.1.89"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atoi"
|
||||
version = "2.0.0"
|
||||
@@ -851,12 +840,9 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
|
||||
name = "pg-instance-handler"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"mockall",
|
||||
"regex",
|
||||
"rstest",
|
||||
"serde",
|
||||
"serde_yaml",
|
||||
"sqlx",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
@@ -1255,19 +1241,6 @@ dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_yaml"
|
||||
version = "0.9.34+deprecated"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
|
||||
dependencies = [
|
||||
"indexmap",
|
||||
"itoa",
|
||||
"ryu",
|
||||
"serde",
|
||||
"unsafe-libyaml",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha1"
|
||||
version = "0.10.6"
|
||||
@@ -1678,21 +1651,9 @@ dependencies = [
|
||||
"mio",
|
||||
"pin-project-lite",
|
||||
"socket2",
|
||||
"tokio-macros",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-stream"
|
||||
version = "0.1.18"
|
||||
@@ -1825,12 +1786,6 @@ version = "0.1.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
|
||||
|
||||
[[package]]
|
||||
name = "unsafe-libyaml"
|
||||
version = "0.2.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
|
||||
|
||||
[[package]]
|
||||
name = "untrusted"
|
||||
version = "0.9.0"
|
||||
|
||||
+1
-4
@@ -4,14 +4,11 @@ version = "0.1.0"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
async-trait = "0.1"
|
||||
regex = "1.12.2"
|
||||
sqlx = { version = "0.8.6", features = ["postgres", "runtime-tokio-rustls"] }
|
||||
tokio = { version = "1.49.0", features = ["rt-multi-thread", "macros"] }
|
||||
tokio = { version = "1.49.0", features = ["rt-multi-thread"] }
|
||||
tracing = "0.1.44"
|
||||
tracing-subscriber = "0.3.22"
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_yaml = "0.9"
|
||||
|
||||
[dev-dependencies]
|
||||
mockall = "0.14.0"
|
||||
|
||||
-23
@@ -1,23 +0,0 @@
|
||||
FROM rust:1.93 AS builder
|
||||
|
||||
RUN rustup target add x86_64-unknown-linux-musl
|
||||
RUN apt-get update && apt-get install -y musl-tools && rm -rf /var/lib/apt/lists/*
|
||||
|
||||
WORKDIR /usr/src/app
|
||||
|
||||
COPY ./Cargo.toml ./Cargo.lock ./
|
||||
COPY ./src ./src
|
||||
|
||||
RUN cargo build --release --target x86_64-unknown-linux-musl
|
||||
|
||||
# Runtime image with postgresql-client tooling available
|
||||
FROM debian:bookworm-slim
|
||||
RUN apt-get update && \
|
||||
apt-get install -y --no-install-recommends postgresql-client ca-certificates && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
COPY --from=builder /usr/src/app/target/x86_64-unknown-linux-musl/release/pg-instance-handler /usr/local/bin/pg-instance-handler
|
||||
|
||||
WORKDIR /
|
||||
|
||||
ENTRYPOINT ["/usr/local/bin/pg-instance-handler"]
|
||||
@@ -1,83 +0,0 @@
|
||||
# pg-instance-handler
|
||||
|
||||
Reconcilie des bases PostgreSQL à partir d’un fichier YAML et gère leur cycle de vie (bases, utilisateurs, droits `CONNECT`, ownership).
|
||||
|
||||
## Sommaire
|
||||
- Aperçu rapide
|
||||
- Démarrage rapide (Docker Compose)
|
||||
- Spécification YAML
|
||||
- Variables d’environnement
|
||||
- Build local et exécution
|
||||
- Opérations et sécurité
|
||||
- Dépannage
|
||||
|
||||
## Aperçu rapide
|
||||
- Crée les bases manquantes et supprime celles non listées (sauf `postgres`).
|
||||
- Crée/actualise un utilisateur par base, lui donne l’accès et l’ownership de la base.
|
||||
- Définition de l’état désiré via un YAML monté dans le conteneur.
|
||||
|
||||
Code pertinent:
|
||||
- Reconciliation: [src/app/database/database_service.rs](src/app/database/database_service.rs)
|
||||
- Spéc YAML / lecture état désiré: [src/actors/driven/database_file_retriever.rs](src/actors/driven/database_file_retriever.rs)
|
||||
- Accès Postgres (sqlx): [src/actors/driven/sqlx_handler.rs](src/actors/driven/sqlx_handler.rs)
|
||||
- Entrée: [src/main.rs](src/main.rs)
|
||||
- Conteneurs: [Dockerfile](Dockerfile), [docker-compose.yml](docker-compose.yml)
|
||||
|
||||
## Démarrage rapide (Docker Compose)
|
||||
1) Préparez votre fichier YAML (exemple ci-dessous) à la racine du projet sous `database.yaml`.
|
||||
2) Lancez:
|
||||
|
||||
```bash
|
||||
docker compose up --build
|
||||
```
|
||||
|
||||
Le service va:
|
||||
- se connecter à PostgreSQL via `DATABASE_URL` (admin),
|
||||
- réconcilier les bases selon `database.yaml`,
|
||||
- créer/mettre à jour les utilisateurs et donner l’ownership,
|
||||
|
||||
## Spécification YAML
|
||||
Exemple minimal: [database.yaml](database.yaml)
|
||||
|
||||
```yaml
|
||||
databases:
|
||||
- name: hello
|
||||
user: hello_user
|
||||
password: hello_password
|
||||
```
|
||||
|
||||
- `name`: nom de la base à gérer
|
||||
- `user`: utilisateur propriétaire/consommateur de la base
|
||||
- `password`: mot de passe de cet utilisateur
|
||||
|
||||
Notes:
|
||||
- Les noms (`name`, `user`) sont validés (lettres, chiffres, underscore; commencent par lettre/underscore).
|
||||
- Le programme ne supprime jamais la base `postgres`.
|
||||
|
||||
## Variables d’environnement
|
||||
- `DATABASE_URL`: URL admin PostgreSQL pour l’orchestrateur (ex: `postgres://postgres:postgres@postgres:5432/postgres`).
|
||||
|
||||
Les exemples d’ENV sont indiqués dans [docker-compose.yml](docker-compose.yml).
|
||||
|
||||
## Build local et exécution
|
||||
```bash
|
||||
cargo build
|
||||
```
|
||||
|
||||
Exécution locale (besoin d’un Postgres accessible et d’un `database.yaml`):
|
||||
```bash
|
||||
export DATABASE_URL="postgres://postgres:postgres@localhost:5432/postgres"
|
||||
|
||||
cargo run
|
||||
```
|
||||
|
||||
## Opérations et sécurité
|
||||
- Suppression: par défaut, toute base non listée est supprimée (sauf `postgres`). Pour un MVP, gardez le YAML strict et versionné. On pourra ajouter un mode « dry-run » et/ou restreindre la gestion à un préfixe (ex: `managed_`).
|
||||
- Permissions: l’utilisateur est `LOGIN PASSWORD`, reçoit `CONNECT` et devient owner de sa base.
|
||||
- Droits DB: `DATABASE_URL` doit pointer une base admin avec droits de création/suppression et gestion des rôles.
|
||||
|
||||
## Dépannage
|
||||
- `DROP DATABASE` échoue: des connexions actives empêchent la suppression. Le programme essaie de terminer les sessions avant `DROP`; réessayez.
|
||||
|
||||
---
|
||||
Contributions bienvenues. Pour toute évolution (dry-run, périmètre géré, migrations), ouvrez une issue/PR.
|
||||
@@ -0,0 +1 @@
|
||||
hello
|
||||
@@ -1,4 +0,0 @@
|
||||
databases:
|
||||
- name: hello
|
||||
user: hello_user
|
||||
password: hello_password
|
||||
@@ -11,15 +11,3 @@ services:
|
||||
interval: 5s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
|
||||
pg-instance-handler:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
environment:
|
||||
DATABASE_URL: postgres://postgres:postgres@postgres:5432/postgres
|
||||
volumes:
|
||||
- ./database.yaml:/database.yaml:ro
|
||||
@@ -1,172 +1,69 @@
|
||||
use crate::app::database::ports::driven::database_port::{
|
||||
ForGettingDatabasesWantedState, GrantDbAccess, WantedState,
|
||||
use std::{
|
||||
fs::File,
|
||||
io::{BufRead, BufReader},
|
||||
};
|
||||
use regex::Regex;
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::app::database_port::ForGettingDatabasesWantedState;
|
||||
pub struct DatabaseFileRetriever {
|
||||
file_path: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
|
||||
pub struct DbSpec {
|
||||
pub name: String,
|
||||
pub user: String,
|
||||
pub password: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct DesiredState {
|
||||
pub databases: Vec<DbSpec>,
|
||||
}
|
||||
|
||||
impl DatabaseFileRetriever {
|
||||
pub fn new(file_path: String) -> Self {
|
||||
DatabaseFileRetriever { file_path }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl ForGettingDatabasesWantedState for DatabaseFileRetriever {
|
||||
async fn get_wanted_state(&self) -> Result<WantedState, String> {
|
||||
let state = load_desired_state(&self.file_path)?;
|
||||
|
||||
let mut databases: Vec<String> = Vec::new();
|
||||
let mut users: Vec<String> = Vec::new();
|
||||
let mut user_passwords: std::collections::BTreeMap<String, String> =
|
||||
std::collections::BTreeMap::new();
|
||||
let mut database_owners: std::collections::BTreeMap<String, String> =
|
||||
std::collections::BTreeMap::new();
|
||||
let mut grants: Vec<GrantDbAccess> = Vec::new();
|
||||
|
||||
let ident_re = Regex::new(r"^[A-Za-z_][A-Za-z0-9_]*$").map_err(|e| e.to_string())?;
|
||||
|
||||
for db in state.databases {
|
||||
if !ident_re.is_match(&db.name) {
|
||||
return Err(format!("invalid database name: '{}'", db.name));
|
||||
}
|
||||
if !ident_re.is_match(&db.user) {
|
||||
return Err(format!("invalid user name: '{}'", db.user));
|
||||
}
|
||||
|
||||
if !databases.contains(&db.name) {
|
||||
databases.push(db.name.clone());
|
||||
}
|
||||
if !users.contains(&db.user) {
|
||||
users.push(db.user.clone());
|
||||
}
|
||||
|
||||
match user_passwords.get(&db.user) {
|
||||
Some(existing) if existing != &db.password => {
|
||||
return Err(format!(
|
||||
"conflicting passwords provided for user '{}'",
|
||||
db.user
|
||||
));
|
||||
}
|
||||
Some(_) => {}
|
||||
None => {
|
||||
user_passwords.insert(db.user.clone(), db.password.clone());
|
||||
}
|
||||
}
|
||||
|
||||
database_owners.insert(db.name.clone(), db.user.clone());
|
||||
|
||||
grants.push(GrantDbAccess {
|
||||
user: db.user,
|
||||
database: db.name,
|
||||
});
|
||||
fn get_wanted_databases(&self) -> Result<Vec<String>, String> {
|
||||
let file = File::open(&self.file_path)
|
||||
.map_err(|e| format!("failed to open '{}': {}", self.file_path, e))?;
|
||||
let reader = BufReader::new(file);
|
||||
let mut databases = Vec::new();
|
||||
for line in reader.lines() {
|
||||
let line = line.map_err(|e| format!("failed to read line: {}", e))?;
|
||||
databases.push(line);
|
||||
}
|
||||
|
||||
Ok(WantedState {
|
||||
databases,
|
||||
users,
|
||||
user_passwords,
|
||||
database_owners,
|
||||
grants,
|
||||
})
|
||||
Ok(databases)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn load_desired_state(path: &str) -> Result<DesiredState, String> {
|
||||
let content =
|
||||
std::fs::read_to_string(path).map_err(|e| format!("failed to read '{}': {}", path, e))?;
|
||||
let state: DesiredState = serde_yaml::from_str(&content)
|
||||
.map_err(|e| format!("failed to parse YAML '{}': {}", path, e))?;
|
||||
Ok(state)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::collections::BTreeMap;
|
||||
use rstest::rstest;
|
||||
use std::io::Write;
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[test]
|
||||
fn wanted_state_keeps_passwords_and_owners() {
|
||||
let tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
std::fs::write(
|
||||
tmp.path(),
|
||||
r#"databases:
|
||||
- name: hello
|
||||
user: hello_user
|
||||
password: secret
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
#[rstest]
|
||||
#[case::two_lines(Some(vec!["db1", "db2"]), Some(vec!["db1".to_string(), "db2".to_string()]))]
|
||||
#[case::empty(Some(vec![]), Some(vec![]))]
|
||||
#[case::missing(None, None)]
|
||||
fn get_wanted_databases_parametrized(
|
||||
#[case] lines: Option<Vec<&str>>,
|
||||
#[case] expected: Option<Vec<String>>,
|
||||
) {
|
||||
let dir = tempdir().unwrap();
|
||||
let file_path = dir.path().join("dbs.txt");
|
||||
|
||||
let retriever = DatabaseFileRetriever::new(tmp.path().to_string_lossy().to_string());
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let wanted = rt.block_on(retriever.get_wanted_state()).unwrap();
|
||||
if let Some(lines) = &lines {
|
||||
let mut file = File::create(&file_path).unwrap();
|
||||
for l in lines {
|
||||
writeln!(file, "{}", l).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(wanted.databases, vec!["hello".to_string()]);
|
||||
assert_eq!(wanted.users, vec!["hello_user".to_string()]);
|
||||
let retriever = DatabaseFileRetriever::new(file_path.to_string_lossy().into_owned());
|
||||
let result = retriever.get_wanted_databases();
|
||||
|
||||
let mut expected_pw = BTreeMap::new();
|
||||
expected_pw.insert("hello_user".to_string(), "secret".to_string());
|
||||
assert_eq!(wanted.user_passwords, expected_pw);
|
||||
|
||||
let mut expected_owners = BTreeMap::new();
|
||||
expected_owners.insert("hello".to_string(), "hello_user".to_string());
|
||||
assert_eq!(wanted.database_owners, expected_owners);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn conflicting_passwords_for_same_user_fails() {
|
||||
let tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
std::fs::write(
|
||||
tmp.path(),
|
||||
r#"databases:
|
||||
- name: db1
|
||||
user: u1
|
||||
password: p1
|
||||
- name: db2
|
||||
user: u1
|
||||
password: p2
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let retriever = DatabaseFileRetriever::new(tmp.path().to_string_lossy().to_string());
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let err = rt.block_on(retriever.get_wanted_state()).unwrap_err();
|
||||
assert!(err.contains("conflicting passwords"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_identifiers_fail_fast() {
|
||||
let tmp = tempfile::NamedTempFile::new().unwrap();
|
||||
std::fs::write(
|
||||
tmp.path(),
|
||||
r#"databases:
|
||||
- name: "bad-name"
|
||||
user: ok_user
|
||||
password: secret
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let retriever = DatabaseFileRetriever::new(tmp.path().to_string_lossy().to_string());
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let err = rt.block_on(retriever.get_wanted_state()).unwrap_err();
|
||||
assert!(err.contains("invalid database name"));
|
||||
match expected {
|
||||
Some(expected_vec) => {
|
||||
let got = result.unwrap();
|
||||
assert_eq!(got, expected_vec);
|
||||
}
|
||||
None => {
|
||||
assert!(result.is_err());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+114
-209
@@ -1,17 +1,22 @@
|
||||
use crate::app::database_port::ForManagingDatabases;
|
||||
use regex::Regex;
|
||||
use sqlx::PgPool;
|
||||
use sqlx::postgres::PgPoolOptions;
|
||||
use tracing::{debug, info};
|
||||
|
||||
use crate::app::database::ports::driven::database_port::{ForManagingDatabases, GrantDbAccess};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SQLXHandler {
|
||||
pool: PgPool,
|
||||
database_url: String,
|
||||
}
|
||||
|
||||
impl SQLXHandler {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
SQLXHandler { pool }
|
||||
pub fn new(database_url: String) -> Self {
|
||||
SQLXHandler { database_url }
|
||||
}
|
||||
|
||||
fn make_runtime() -> Result<tokio::runtime::Runtime, String> {
|
||||
tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.map_err(|e| format!("failed to build runtime: {}", e))
|
||||
}
|
||||
|
||||
fn sanitize_identifier(name: &str) -> Result<String, String> {
|
||||
@@ -24,228 +29,128 @@ impl SQLXHandler {
|
||||
Err(format!("invalid database name: '{}'", name))
|
||||
}
|
||||
}
|
||||
|
||||
async fn user_exists(&self, user_ident: &str) -> Result<bool, String> {
|
||||
let exists: Option<bool> =
|
||||
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname = $1)")
|
||||
.bind(user_ident)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("role exists check failed: {}", e))?;
|
||||
Ok(exists.unwrap_or(false))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl ForManagingDatabases for SQLXHandler {
|
||||
async fn list_databases(&self) -> Result<Vec<String>, String> {
|
||||
fn list_databases(&self) -> Result<Vec<String>, String> {
|
||||
info!("Listing PostgreSQL databases");
|
||||
let dbs: Vec<String> = sqlx::query_scalar(
|
||||
r#"SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname"#,
|
||||
)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("list databases failed: {}", e))?;
|
||||
let rt = Self::make_runtime()?;
|
||||
rt.block_on(async {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.connect(&self.database_url)
|
||||
.await
|
||||
.map_err(|e| format!("failed to connect: {}", e))?;
|
||||
|
||||
debug!(?dbs, "Got databases");
|
||||
Ok(dbs)
|
||||
let dbs: Vec<String> = sqlx::query_scalar(
|
||||
r#"SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname"#,
|
||||
)
|
||||
.fetch_all(&pool)
|
||||
.await
|
||||
.map_err(|e| format!("list databases failed: {}", e))?;
|
||||
|
||||
debug!(?dbs, "Got databases");
|
||||
Ok(dbs)
|
||||
})
|
||||
}
|
||||
|
||||
async fn create_database(&self, name: &str) -> Result<(), String> {
|
||||
fn create_database(&self, name: &str) -> Result<(), String> {
|
||||
let ident = Self::sanitize_identifier(name)?;
|
||||
info!(database = %ident, "Creating database");
|
||||
let sql = format!("CREATE DATABASE {}", ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("create database failed: {}", e))?;
|
||||
let rt = Self::make_runtime()?;
|
||||
rt.block_on(async {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.connect(&self.database_url)
|
||||
.await
|
||||
.map_err(|e| format!("failed to connect: {}", e))?;
|
||||
|
||||
info!(database = %ident, "Database created");
|
||||
|
||||
Ok(())
|
||||
let sql = format!("CREATE DATABASE {}", ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.map_err(|e| format!("create database failed: {}", e))?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
async fn delete_database(&self, name: &str) -> Result<(), String> {
|
||||
fn delete_database(&self, name: &str) -> Result<(), String> {
|
||||
let ident = Self::sanitize_identifier(name)?;
|
||||
info!(database = %ident, "Deleting database");
|
||||
// Terminate connections before dropping (best effort)
|
||||
let terminate_sql = r#"
|
||||
let rt = Self::make_runtime()?;
|
||||
rt.block_on(async {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.connect(&self.database_url)
|
||||
.await
|
||||
.map_err(|e| format!("failed to connect: {}", e))?;
|
||||
|
||||
// Terminate connections before dropping (best effort)
|
||||
let terminate_sql = r#"
|
||||
SELECT pg_terminate_backend(pid)
|
||||
FROM pg_stat_activity
|
||||
WHERE datname = $1 AND pid <> pg_backend_pid()
|
||||
"#;
|
||||
let _ = sqlx::query(terminate_sql)
|
||||
.bind(&ident)
|
||||
.execute(&self.pool)
|
||||
.await;
|
||||
let _ = sqlx::query(terminate_sql).bind(&ident).execute(&pool).await;
|
||||
|
||||
let sql = format!("DROP DATABASE IF EXISTS {}", ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("drop database failed: {}", e))?;
|
||||
|
||||
info!(database = %ident, "Database deleted");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_users(&self) -> Result<Vec<String>, String> {
|
||||
info!("Listing PostgreSQL users");
|
||||
let users: Vec<String> = sqlx::query_scalar(
|
||||
r#"SELECT rolname FROM pg_roles WHERE rolcanlogin = true ORDER BY rolname"#,
|
||||
)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("list users failed: {}", e))?;
|
||||
|
||||
debug!(?users, "Got users");
|
||||
Ok(users)
|
||||
}
|
||||
|
||||
async fn list_grants(&self) -> Result<Vec<GrantDbAccess>, String> {
|
||||
info!("Listing explicit database grants");
|
||||
// We only return explicit CONNECT grants from database ACLs (datacl).
|
||||
// This avoids exploding results when PUBLIC has CONNECT.
|
||||
let rows: Vec<(String, Option<String>)> = sqlx::query_as(
|
||||
r#"SELECT datname, array_to_string(datacl, ',') AS acl
|
||||
FROM pg_database
|
||||
WHERE datistemplate = false
|
||||
ORDER BY datname"#,
|
||||
)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("list grants failed: {}", e))?;
|
||||
|
||||
let mut grants: Vec<GrantDbAccess> = Vec::new();
|
||||
for (db_name, acl_opt) in rows {
|
||||
let Some(acl) = acl_opt else {
|
||||
continue;
|
||||
};
|
||||
for entry in acl.split(',') {
|
||||
// Entry format: grantee=privs/grantor
|
||||
// Example: hello_user=CTc/postgres
|
||||
let (lhs, rhs) = match entry.split_once('=') {
|
||||
Some(v) => v,
|
||||
None => continue,
|
||||
};
|
||||
let grantee = lhs.trim();
|
||||
if grantee.is_empty() {
|
||||
// PUBLIC
|
||||
continue;
|
||||
}
|
||||
let privs = rhs.split('/').next().unwrap_or("");
|
||||
if privs.contains('c') {
|
||||
grants.push(GrantDbAccess {
|
||||
user: grantee.to_string(),
|
||||
database: db_name.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
debug!(?grants, "Got grants");
|
||||
Ok(grants)
|
||||
}
|
||||
|
||||
async fn ensure_user(&self, user: &str, password: &str) -> Result<(), String> {
|
||||
let user_ident = Self::sanitize_identifier(user)?;
|
||||
info!(user = %user_ident, "Ensuring user (password + login)");
|
||||
|
||||
let quoted_password: String = sqlx::query_scalar("SELECT quote_literal($1)")
|
||||
.bind(password)
|
||||
.fetch_one(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("failed to quote password: {}", e))?;
|
||||
|
||||
let exists = self.user_exists(&user_ident).await?;
|
||||
if exists {
|
||||
let sql = format!(
|
||||
"ALTER ROLE {} WITH LOGIN PASSWORD {} NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION",
|
||||
user_ident, quoted_password
|
||||
);
|
||||
let sql = format!("DROP DATABASE IF EXISTS {}", ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.map_err(|e| format!("alter role failed: {}", e))?;
|
||||
} else {
|
||||
let sql = format!(
|
||||
"CREATE ROLE {} WITH LOGIN PASSWORD {} NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION",
|
||||
user_ident, quoted_password
|
||||
);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("create role failed: {}", e))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_user(&self, name: &str) -> Result<(), String> {
|
||||
let user_ident = Self::sanitize_identifier(name)?;
|
||||
info!(user = %user_ident, "Deleting user");
|
||||
// Best-effort cleanup to avoid DROP ROLE failures.
|
||||
let reassign = format!("REASSIGN OWNED BY {} TO postgres", user_ident);
|
||||
let _ = sqlx::query(&reassign).execute(&self.pool).await;
|
||||
|
||||
let drop_owned = format!("DROP OWNED BY {}", user_ident);
|
||||
let _ = sqlx::query(&drop_owned).execute(&self.pool).await;
|
||||
|
||||
let sql = format!("DROP ROLE IF EXISTS {}", user_ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("delete user failed: {}", e))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn revoke_public_connect(&self, database: &str) -> Result<(), String> {
|
||||
let db_ident = Self::sanitize_identifier(database)?;
|
||||
info!(database = %db_ident, "Revoking CONNECT from PUBLIC");
|
||||
let revoke = format!("REVOKE CONNECT ON DATABASE {} FROM PUBLIC", db_ident);
|
||||
sqlx::query(&revoke)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("revoke public connect failed: {}", e))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn ensure_db_owner(&self, database: &str, owner: &str) -> Result<(), String> {
|
||||
let db_ident = Self::sanitize_identifier(database)?;
|
||||
let owner_ident = Self::sanitize_identifier(owner)?;
|
||||
info!(database = %db_ident, owner = %owner_ident, "Ensuring database owner");
|
||||
let sql = format!("ALTER DATABASE {} OWNER TO {}", db_ident, owner_ident);
|
||||
sqlx::query(&sql)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("alter owner failed: {}", e))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn grant_access(&self, user: &str, database: &str) -> Result<(), String> {
|
||||
let db_ident = Self::sanitize_identifier(database)?;
|
||||
let user_ident = Self::sanitize_identifier(user)?;
|
||||
info!(user = %user_ident, database = %db_ident, "Granting access");
|
||||
let grant = format!("GRANT CONNECT ON DATABASE {} TO {}", db_ident, user_ident);
|
||||
sqlx::query(&grant)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("grant access failed: {}", e))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn revoke_access(&self, user: &str, database: &str) -> Result<(), String> {
|
||||
let db_ident = Self::sanitize_identifier(database)?;
|
||||
let user_ident = Self::sanitize_identifier(user)?;
|
||||
info!(user = %user_ident, database = %db_ident, "Revoking access");
|
||||
let revoke = format!(
|
||||
"REVOKE CONNECT ON DATABASE {} FROM {}",
|
||||
db_ident, user_ident
|
||||
);
|
||||
sqlx::query(&revoke)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(|e| format!("revoke access failed: {}", e))?;
|
||||
Ok(())
|
||||
.map_err(|e| format!("drop database failed: {}", e))?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn sanitize_identifier_allows_valid() {
|
||||
for ok in ["db", "_db", "db_123", "A_b1"] {
|
||||
assert!(
|
||||
SQLXHandler::sanitize_identifier(ok).is_ok(),
|
||||
"{} should be valid",
|
||||
ok
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sanitize_identifier_rejects_invalid() {
|
||||
for bad in ["", "1db", "db-1", "db name", "db;drop", "db$", "db."] {
|
||||
assert!(
|
||||
SQLXHandler::sanitize_identifier(bad).is_err(),
|
||||
"{} should be invalid",
|
||||
bad
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Optional smoke test against a real Postgres instance.
|
||||
// Set TEST_ADMIN_DATABASE_URL to something like:
|
||||
// postgres://postgres:postgres@localhost/postgres
|
||||
// Run with: cargo test -- --ignored
|
||||
#[test]
|
||||
#[ignore]
|
||||
fn postgres_smoke_create_list_delete() {
|
||||
let url = std::env::var("TEST_ADMIN_DATABASE_URL")
|
||||
.expect("set TEST_ADMIN_DATABASE_URL to an admin database URL");
|
||||
let handler = SQLXHandler::new(url);
|
||||
|
||||
let db_name = format!("testdb_{}", std::process::id());
|
||||
|
||||
// Ensure clean slate (ignore errors)
|
||||
let _ = handler.delete_database(&db_name);
|
||||
|
||||
handler.create_database(&db_name).expect("create ok");
|
||||
let list = handler.list_databases().expect("list ok");
|
||||
assert!(list.contains(&db_name));
|
||||
|
||||
handler.delete_database(&db_name).expect("drop ok");
|
||||
let list2 = handler.list_databases().expect("list ok");
|
||||
assert!(!list2.contains(&db_name));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,410 +0,0 @@
|
||||
use tracing::info;
|
||||
|
||||
use crate::app::database::ports::driven::database_port::{
|
||||
ForGettingDatabasesWantedState, ForManagingDatabases,
|
||||
};
|
||||
|
||||
pub struct DatabaseService<X, Y>
|
||||
where
|
||||
X: ForManagingDatabases,
|
||||
Y: ForGettingDatabasesWantedState,
|
||||
{
|
||||
database_handler: X,
|
||||
wanted_state_handler: Y,
|
||||
}
|
||||
|
||||
impl<X, Y> DatabaseService<X, Y>
|
||||
where
|
||||
X: ForManagingDatabases,
|
||||
Y: ForGettingDatabasesWantedState,
|
||||
{
|
||||
pub fn new(database_handler: X, wanted_state_handler: Y) -> Self {
|
||||
DatabaseService {
|
||||
database_handler,
|
||||
wanted_state_handler,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn reconcile_databases(&self) -> Result<(), String> {
|
||||
let wanted_state = self.wanted_state_handler.get_wanted_state().await?;
|
||||
let existing_databases = self.database_handler.list_databases().await?;
|
||||
let existing_users = self.database_handler.list_users().await?;
|
||||
let existing_grants = self.database_handler.list_grants().await?;
|
||||
|
||||
info!(
|
||||
"Reconciling databases. Wanted: {:?}, Existing: {:?}",
|
||||
wanted_state.databases, existing_databases
|
||||
);
|
||||
|
||||
for db in &wanted_state.databases {
|
||||
if !existing_databases.contains(db) {
|
||||
self.database_handler.create_database(db).await?;
|
||||
}
|
||||
}
|
||||
|
||||
for db in &existing_databases {
|
||||
if !wanted_state.databases.contains(db) && db != "postgres" {
|
||||
self.database_handler.delete_database(db).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure users exist and have expected password.
|
||||
for user in &wanted_state.users {
|
||||
if user == "postgres" {
|
||||
continue;
|
||||
}
|
||||
let password = wanted_state
|
||||
.user_passwords
|
||||
.get(user)
|
||||
.ok_or_else(|| format!("missing password for user '{}'", user))?;
|
||||
self.database_handler.ensure_user(user, password).await?;
|
||||
}
|
||||
|
||||
for user in &existing_users {
|
||||
if !wanted_state.users.contains(user) && user != "postgres" {
|
||||
self.database_handler.delete_user(user).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// Enforce connect policy: remove PUBLIC and ensure expected ownership.
|
||||
// PostgreSQL grants CONNECT on databases to PUBLIC by default. If you want users to only
|
||||
// connect to their declared databases, we must revoke it at least on `postgres`.
|
||||
self.database_handler
|
||||
.revoke_public_connect("postgres")
|
||||
.await?;
|
||||
|
||||
for db in &wanted_state.databases {
|
||||
if db == "postgres" {
|
||||
continue;
|
||||
}
|
||||
self.database_handler.revoke_public_connect(db).await?;
|
||||
}
|
||||
|
||||
for (db, owner) in &wanted_state.database_owners {
|
||||
if db == "postgres" {
|
||||
continue;
|
||||
}
|
||||
self.database_handler.ensure_db_owner(db, owner).await?;
|
||||
}
|
||||
|
||||
for grant in &wanted_state.grants {
|
||||
let already_granted = existing_grants
|
||||
.iter()
|
||||
.any(|g| g.user == grant.user && g.database == grant.database);
|
||||
if !already_granted {
|
||||
self.database_handler
|
||||
.grant_access(&grant.user, &grant.database)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
// Revoke explicit grants on databases we manage when they are no longer wanted.
|
||||
for existing_grant in &existing_grants {
|
||||
if !wanted_state.databases.contains(&existing_grant.database) {
|
||||
continue;
|
||||
}
|
||||
let still_wanted = wanted_state
|
||||
.grants
|
||||
.iter()
|
||||
.any(|g| g.user == existing_grant.user && g.database == existing_grant.database);
|
||||
if !still_wanted {
|
||||
self.database_handler
|
||||
.revoke_access(&existing_grant.user, &existing_grant.database)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::app::database::ports::driven::database_port::{
|
||||
MockForGettingDatabasesWantedState, MockForManagingDatabases, WantedState,
|
||||
};
|
||||
use mockall::predicate::*;
|
||||
use rstest::rstest;
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
#[rstest]
|
||||
// No databases exist and none are wanted
|
||||
#[case::no_existing_no_wanted(Ok(vec![]), Ok(vec![]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// No databases exist but one is wanted
|
||||
#[case::no_existing_one_wanted(Ok(vec![]), Ok(vec!["db1".to_string()]), vec!["db1".to_string()], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// A database exists and is wanted
|
||||
#[case::exists_and_wanted(Ok(vec!["db1".to_string()]), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// One database exists and another is wanted
|
||||
#[case::different_existing_vs_wanted(Ok(vec!["db1".to_string()]), Ok(vec!["db2".to_string()]), vec!["db2".to_string()], Ok(()), vec!["db1".to_string()], Ok(()), Ok(()))]
|
||||
// A database exists but is not wanted
|
||||
#[case::existing_not_wanted(Ok(vec!["db1".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string()], Ok(()), Ok(()))]
|
||||
// Multiple databases exist and none are wanted
|
||||
#[case::multiple_existing_none_wanted(Ok(vec!["db1".to_string(), "db2".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string(), "db2".to_string()], Ok(()), Ok(()))]
|
||||
// No databases exist but multiple are wanted
|
||||
#[case::no_existing_multiple_wanted(Ok(vec![]), Ok(vec!["db1".to_string(), "db2".to_string()]), vec!["db1".to_string(), "db2".to_string()], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// Multiple databases exist and some are wanted
|
||||
#[case::multiple_existing_some_wanted(Ok(vec!["db1".to_string(), "db2".to_string(), "db3".to_string()]), Ok(vec!["db2".to_string(), "db4".to_string()]), vec!["db4".to_string()], Ok(()), vec!["db1".to_string(), "db3".to_string()], Ok(()), Ok(()))]
|
||||
// Error retrieving existing databases
|
||||
#[case::error_list_databases(Err("DB Error".to_string()), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Err("DB Error".to_string()))]
|
||||
// Error retrieving wanted databases
|
||||
#[case::error_wanted_databases(Ok(vec!["db1".to_string()]), Err("Wanted State Error".to_string()), vec![], Ok(()), vec![], Ok(()), Err("Wanted State Error".to_string()))]
|
||||
// Error creating a database
|
||||
#[case::error_create_database(Ok(vec![]), Ok(vec!["db1".to_string()]), vec!["db1".to_string()], Err("Creation Error".to_string()), vec![], Ok(()), Err("Creation Error".to_string()))]
|
||||
// Error deleting a database
|
||||
#[case::error_delete_database(Ok(vec!["db1".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string()], Err("Deletion Error".to_string()), Err("Deletion Error".to_string()))]
|
||||
// If database name is "postgres", delete should not be called
|
||||
#[case::dont_delete_postgres(Ok(vec!["postgres".to_string(), "db1".to_string()]), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
fn test_reconcile_databases(
|
||||
#[case] existing_dbs_result: Result<Vec<String>, String>,
|
||||
#[case] wanted_dbs_result: Result<Vec<String>, String>,
|
||||
#[case] expected_database_creation_param: Vec<String>,
|
||||
#[case] expected_database_creation_result: Result<(), String>,
|
||||
#[case] expected_database_deletion_param: Vec<String>,
|
||||
#[case] expected_database_deletion_result: Result<(), String>,
|
||||
#[case] expected_result: Result<(), String>,
|
||||
) {
|
||||
let mut mock_db_handler = MockForManagingDatabases::new();
|
||||
let mut mock_wanted_state_handler = MockForGettingDatabasesWantedState::new();
|
||||
|
||||
mock_wanted_state_handler
|
||||
.expect_get_wanted_state()
|
||||
.returning(move || {
|
||||
wanted_dbs_result.clone().map(|dbs| WantedState {
|
||||
databases: dbs,
|
||||
users: vec![],
|
||||
user_passwords: BTreeMap::new(),
|
||||
database_owners: BTreeMap::new(),
|
||||
grants: vec![],
|
||||
})
|
||||
});
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_databases()
|
||||
.returning(move || existing_dbs_result.clone());
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_users()
|
||||
.times(0..)
|
||||
.returning(|| Ok(vec![]));
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_grants()
|
||||
.times(0..)
|
||||
.returning(|| Ok(vec![]));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.times(0..)
|
||||
.returning(|_| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_db_owner()
|
||||
.times(0..)
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_user()
|
||||
.times(0..)
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
for db_name in expected_database_creation_param.clone() {
|
||||
let db_name_clone = db_name.clone();
|
||||
let expected_database_creation_result = expected_database_creation_result.clone();
|
||||
mock_db_handler
|
||||
.expect_create_database()
|
||||
.with(eq(db_name_clone))
|
||||
.returning(move |_| expected_database_creation_result.clone());
|
||||
}
|
||||
|
||||
for db_name in expected_database_deletion_param.clone() {
|
||||
let db_name_clone = db_name.clone();
|
||||
let expected_database_deletion_result = expected_database_deletion_result.clone();
|
||||
mock_db_handler
|
||||
.expect_delete_database()
|
||||
.with(eq(db_name_clone))
|
||||
.returning(move |_| expected_database_deletion_result.clone());
|
||||
}
|
||||
|
||||
let service = DatabaseService::new(mock_db_handler, mock_wanted_state_handler);
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let result = rt.block_on(service.reconcile_databases());
|
||||
assert_eq!(result, expected_result);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ensure_user_is_called_with_password() {
|
||||
let mut mock_db_handler = MockForManagingDatabases::new();
|
||||
let mut mock_wanted_state_handler = MockForGettingDatabasesWantedState::new();
|
||||
|
||||
mock_wanted_state_handler
|
||||
.expect_get_wanted_state()
|
||||
.returning(|| {
|
||||
let mut user_passwords = BTreeMap::new();
|
||||
user_passwords.insert("u1".to_string(), "p1".to_string());
|
||||
|
||||
Ok(WantedState {
|
||||
databases: vec![],
|
||||
users: vec!["u1".to_string()],
|
||||
user_passwords,
|
||||
database_owners: BTreeMap::new(),
|
||||
grants: vec![],
|
||||
})
|
||||
});
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_databases()
|
||||
.returning(|| Ok(vec![]));
|
||||
mock_db_handler.expect_list_users().returning(|| Ok(vec![]));
|
||||
mock_db_handler
|
||||
.expect_list_grants()
|
||||
.returning(|| Ok(vec![]));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_user()
|
||||
.with(eq("u1"), eq("p1"))
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.with(eq("postgres"))
|
||||
.times(1)
|
||||
.returning(|_| Ok(()));
|
||||
mock_db_handler
|
||||
.expect_ensure_db_owner()
|
||||
.times(0)
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
let service = DatabaseService::new(mock_db_handler, mock_wanted_state_handler);
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(service.reconcile_databases()).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn enforces_public_policy_ownership_and_grants() {
|
||||
let mut mock_db_handler = MockForManagingDatabases::new();
|
||||
let mut mock_wanted_state_handler = MockForGettingDatabasesWantedState::new();
|
||||
|
||||
mock_wanted_state_handler
|
||||
.expect_get_wanted_state()
|
||||
.returning(|| {
|
||||
let mut user_passwords = BTreeMap::new();
|
||||
user_passwords.insert("u1".to_string(), "p1".to_string());
|
||||
let mut database_owners = BTreeMap::new();
|
||||
database_owners.insert("db1".to_string(), "u1".to_string());
|
||||
|
||||
Ok(WantedState {
|
||||
databases: vec!["db1".to_string()],
|
||||
users: vec!["u1".to_string()],
|
||||
user_passwords,
|
||||
database_owners,
|
||||
grants: vec![
|
||||
crate::app::database::ports::driven::database_port::GrantDbAccess {
|
||||
user: "u1".to_string(),
|
||||
database: "db1".to_string(),
|
||||
},
|
||||
],
|
||||
})
|
||||
});
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_databases()
|
||||
.returning(|| Ok(vec!["db1".to_string()]));
|
||||
mock_db_handler
|
||||
.expect_list_users()
|
||||
.returning(|| Ok(vec!["u1".to_string()]));
|
||||
mock_db_handler
|
||||
.expect_list_grants()
|
||||
.returning(|| Ok(vec![]));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_user()
|
||||
.with(eq("u1"), eq("p1"))
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.with(eq("postgres"))
|
||||
.times(1)
|
||||
.returning(|_| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.with(eq("db1"))
|
||||
.returning(|_| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_db_owner()
|
||||
.with(eq("db1"), eq("u1"))
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_grant_access()
|
||||
.with(eq("u1"), eq("db1"))
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
let service = DatabaseService::new(mock_db_handler, mock_wanted_state_handler);
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(service.reconcile_databases()).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn revokes_stale_grants_on_managed_databases() {
|
||||
let mut mock_db_handler = MockForManagingDatabases::new();
|
||||
let mut mock_wanted_state_handler = MockForGettingDatabasesWantedState::new();
|
||||
|
||||
mock_wanted_state_handler
|
||||
.expect_get_wanted_state()
|
||||
.returning(|| {
|
||||
Ok(WantedState {
|
||||
databases: vec!["db1".to_string()],
|
||||
users: vec![],
|
||||
user_passwords: BTreeMap::new(),
|
||||
database_owners: BTreeMap::new(),
|
||||
grants: vec![],
|
||||
})
|
||||
});
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_databases()
|
||||
.returning(|| Ok(vec!["db1".to_string()]));
|
||||
mock_db_handler.expect_list_users().returning(|| Ok(vec![]));
|
||||
mock_db_handler.expect_list_grants().returning(|| {
|
||||
Ok(vec![
|
||||
crate::app::database::ports::driven::database_port::GrantDbAccess {
|
||||
user: "u2".to_string(),
|
||||
database: "db1".to_string(),
|
||||
},
|
||||
])
|
||||
});
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.with(eq("postgres"))
|
||||
.times(1)
|
||||
.returning(|_| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_public_connect()
|
||||
.with(eq("db1"))
|
||||
.returning(|_| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_revoke_access()
|
||||
.with(eq("u2"), eq("db1"))
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
mock_db_handler
|
||||
.expect_ensure_db_owner()
|
||||
.times(0)
|
||||
.returning(|_, _| Ok(()));
|
||||
mock_db_handler
|
||||
.expect_ensure_user()
|
||||
.times(0)
|
||||
.returning(|_, _| Ok(()));
|
||||
|
||||
let service = DatabaseService::new(mock_db_handler, mock_wanted_state_handler);
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(service.reconcile_databases()).unwrap();
|
||||
}
|
||||
}
|
||||
@@ -1,40 +0,0 @@
|
||||
#[cfg_attr(test, mockall::automock)]
|
||||
#[async_trait::async_trait]
|
||||
pub trait ForGettingDatabasesWantedState {
|
||||
async fn get_wanted_state(&self) -> Result<WantedState, String>;
|
||||
}
|
||||
|
||||
#[cfg_attr(test, mockall::automock)]
|
||||
#[async_trait::async_trait]
|
||||
pub trait ForManagingDatabases {
|
||||
async fn list_databases(&self) -> Result<Vec<String>, String>;
|
||||
async fn list_users(&self) -> Result<Vec<String>, String>;
|
||||
async fn list_grants(&self) -> Result<Vec<GrantDbAccess>, String>;
|
||||
|
||||
async fn create_database(&self, name: &str) -> Result<(), String>;
|
||||
async fn delete_database(&self, name: &str) -> Result<(), String>;
|
||||
|
||||
async fn ensure_user(&self, user: &str, password: &str) -> Result<(), String>;
|
||||
async fn delete_user(&self, name: &str) -> Result<(), String>;
|
||||
|
||||
async fn revoke_public_connect(&self, database: &str) -> Result<(), String>;
|
||||
async fn ensure_db_owner(&self, database: &str, owner: &str) -> Result<(), String>;
|
||||
|
||||
async fn grant_access(&self, user: &str, database: &str) -> Result<(), String>;
|
||||
async fn revoke_access(&self, user: &str, database: &str) -> Result<(), String>;
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct WantedState {
|
||||
pub databases: Vec<String>,
|
||||
pub users: Vec<String>,
|
||||
pub user_passwords: std::collections::BTreeMap<String, String>,
|
||||
pub database_owners: std::collections::BTreeMap<String, String>,
|
||||
pub grants: Vec<GrantDbAccess>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct GrantDbAccess {
|
||||
pub user: String,
|
||||
pub database: String,
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
#[cfg_attr(test, mockall::automock)]
|
||||
pub trait ForGettingDatabasesWantedState {
|
||||
fn get_wanted_databases(&self) -> Result<Vec<String>, String>;
|
||||
}
|
||||
|
||||
#[cfg_attr(test, mockall::automock)]
|
||||
pub trait ForManagingDatabases {
|
||||
fn list_databases(&self) -> Result<Vec<String>, String>;
|
||||
fn create_database(&self, name: &str) -> Result<(), String>;
|
||||
fn delete_database(&self, name: &str) -> Result<(), String>;
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
use tracing::info;
|
||||
|
||||
use crate::app::database_port::{ForGettingDatabasesWantedState, ForManagingDatabases};
|
||||
|
||||
pub struct DatabaseService<X, Y>
|
||||
where
|
||||
X: ForManagingDatabases,
|
||||
Y: ForGettingDatabasesWantedState,
|
||||
{
|
||||
database_handler: X,
|
||||
wanted_state_handler: Y,
|
||||
}
|
||||
|
||||
impl<X, Y> DatabaseService<X, Y>
|
||||
where
|
||||
X: ForManagingDatabases,
|
||||
Y: ForGettingDatabasesWantedState,
|
||||
{
|
||||
pub fn new(database_handler: X, wanted_state_handler: Y) -> Self {
|
||||
DatabaseService {
|
||||
database_handler,
|
||||
wanted_state_handler,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn reconcile_databases(&self) -> Result<(), String> {
|
||||
let wanted_databases = self.wanted_state_handler.get_wanted_databases()?;
|
||||
let existing_databases = self.database_handler.list_databases()?;
|
||||
|
||||
info!(
|
||||
"Reconciling databases. Wanted: {:?}, Existing: {:?}",
|
||||
wanted_databases, existing_databases
|
||||
);
|
||||
|
||||
for db in &wanted_databases {
|
||||
if !existing_databases.contains(db) {
|
||||
self.database_handler.create_database(db)?;
|
||||
}
|
||||
}
|
||||
|
||||
for db in &existing_databases {
|
||||
if !wanted_databases.contains(db) && db != "postgres" {
|
||||
self.database_handler.delete_database(db)?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::app::database_port::{MockForGettingDatabasesWantedState, MockForManagingDatabases};
|
||||
use mockall::predicate::*;
|
||||
use rstest::rstest;
|
||||
|
||||
#[rstest]
|
||||
// No databases exist and none are wanted
|
||||
#[case::no_existing_no_wanted(Ok(vec![]), Ok(vec![]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// No databases exist but one is wanted
|
||||
#[case::no_existing_one_wanted(Ok(vec![]), Ok(vec!["db1".to_string()]), vec!["db1".to_string()], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// A database exists and is wanted
|
||||
#[case::exists_and_wanted(Ok(vec!["db1".to_string()]), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// One database exists and another is wanted
|
||||
#[case::different_existing_vs_wanted(Ok(vec!["db1".to_string()]), Ok(vec!["db2".to_string()]), vec!["db2".to_string()], Ok(()), vec!["db1".to_string()], Ok(()), Ok(()))]
|
||||
// A database exists but is not wanted
|
||||
#[case::existing_not_wanted(Ok(vec!["db1".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string()], Ok(()), Ok(()))]
|
||||
// Multiple databases exist and none are wanted
|
||||
#[case::multiple_existing_none_wanted(Ok(vec!["db1".to_string(), "db2".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string(), "db2".to_string()], Ok(()), Ok(()))]
|
||||
// No databases exist but multiple are wanted
|
||||
#[case::no_existing_multiple_wanted(Ok(vec![]), Ok(vec!["db1".to_string(), "db2".to_string()]), vec!["db1".to_string(), "db2".to_string()], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
// Multiple databases exist and some are wanted
|
||||
#[case::multiple_existing_some_wanted(Ok(vec!["db1".to_string(), "db2".to_string(), "db3".to_string()]), Ok(vec!["db2".to_string(), "db4".to_string()]), vec!["db4".to_string()], Ok(()), vec!["db1".to_string(), "db3".to_string()], Ok(()), Ok(()))]
|
||||
// Error retrieving existing databases
|
||||
#[case::error_list_databases(Err("DB Error".to_string()), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Err("DB Error".to_string()))]
|
||||
// Error retrieving wanted databases
|
||||
#[case::error_wanted_databases(Ok(vec!["db1".to_string()]), Err("Wanted State Error".to_string()), vec![], Ok(()), vec![], Ok(()), Err("Wanted State Error".to_string()))]
|
||||
// Error creating a database
|
||||
#[case::error_create_database(Ok(vec![]), Ok(vec!["db1".to_string()]), vec!["db1".to_string()], Err("Creation Error".to_string()), vec![], Ok(()), Err("Creation Error".to_string()))]
|
||||
// Error deleting a database
|
||||
#[case::error_delete_database(Ok(vec!["db1".to_string()]), Ok(vec![]), vec![], Ok(()), vec!["db1".to_string()], Err("Deletion Error".to_string()), Err("Deletion Error".to_string()))]
|
||||
// If database name is "postgres", delete should not be called
|
||||
#[case::dont_delete_postgres(Ok(vec!["postgres".to_string(), "db1".to_string()]), Ok(vec!["db1".to_string()]), vec![], Ok(()), vec![], Ok(()), Ok(()))]
|
||||
fn test_reconcile_databases(
|
||||
#[case] existing_dbs_result: Result<Vec<String>, String>,
|
||||
#[case] wanted_dbs_result: Result<Vec<String>, String>,
|
||||
#[case] expected_database_creation_param: Vec<String>,
|
||||
#[case] expected_database_creation_result: Result<(), String>,
|
||||
#[case] expected_database_deletion_param: Vec<String>,
|
||||
#[case] expected_database_deletion_result: Result<(), String>,
|
||||
#[case] expected_result: Result<(), String>,
|
||||
) {
|
||||
let mut mock_db_handler = MockForManagingDatabases::new();
|
||||
let mut mock_wanted_state_handler = MockForGettingDatabasesWantedState::new();
|
||||
|
||||
mock_wanted_state_handler
|
||||
.expect_get_wanted_databases()
|
||||
.returning(move || wanted_dbs_result.clone());
|
||||
|
||||
mock_db_handler
|
||||
.expect_list_databases()
|
||||
.returning(move || existing_dbs_result.clone());
|
||||
|
||||
for db_name in expected_database_creation_param.clone() {
|
||||
let db_name_clone = db_name.clone();
|
||||
let expected_database_creation_result = expected_database_creation_result.clone();
|
||||
mock_db_handler
|
||||
.expect_create_database()
|
||||
.with(eq(db_name_clone))
|
||||
.returning(move |_| expected_database_creation_result.clone());
|
||||
}
|
||||
|
||||
for db_name in expected_database_deletion_param.clone() {
|
||||
let db_name_clone = db_name.clone();
|
||||
let expected_database_deletion_result = expected_database_deletion_result.clone();
|
||||
mock_db_handler
|
||||
.expect_delete_database()
|
||||
.with(eq(db_name_clone))
|
||||
.returning(move |_| expected_database_deletion_result.clone());
|
||||
}
|
||||
|
||||
let service = DatabaseService::new(mock_db_handler, mock_wanted_state_handler);
|
||||
let result = service.reconcile_databases();
|
||||
assert_eq!(result, expected_result);
|
||||
}
|
||||
}
|
||||
+7
-24
@@ -1,12 +1,6 @@
|
||||
mod app {
|
||||
pub mod database {
|
||||
pub mod database_service;
|
||||
pub mod ports {
|
||||
pub mod driven {
|
||||
pub mod database_port;
|
||||
}
|
||||
}
|
||||
}
|
||||
pub mod database_port;
|
||||
pub mod database_service;
|
||||
}
|
||||
|
||||
mod actors {
|
||||
@@ -16,27 +10,16 @@ mod actors {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
fn main() {
|
||||
tracing_subscriber::fmt::init();
|
||||
|
||||
let database_url = std::env::var("DATABASE_URL")
|
||||
.unwrap_or_else(|_| "postgres://postgres:postgres@localhost/postgres".to_string());
|
||||
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.connect(&database_url)
|
||||
.await
|
||||
.expect("failed to connect to postgres");
|
||||
let database_handler = actors::driven::sqlx_handler::SQLXHandler::new(pool);
|
||||
|
||||
let database_handler = actors::driven::sqlx_handler::SQLXHandler::new(database_url);
|
||||
let file_retriever = actors::driven::database_file_retriever::DatabaseFileRetriever::new(
|
||||
"database.yaml".to_string(),
|
||||
"database.txt".to_string(),
|
||||
);
|
||||
|
||||
let database_service = app::database::database_service::DatabaseService::new(
|
||||
database_handler.clone(),
|
||||
file_retriever,
|
||||
);
|
||||
database_service.reconcile_databases().await.unwrap();
|
||||
let service = app::database_service::DatabaseService::new(database_handler, file_retriever);
|
||||
service.reconcile_databases().unwrap();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user