pgroles_operator/
crd.rs

1//! Custom Resource Definition for `PostgresPolicy`.
2//!
3//! Defines the `pgroles.io/v1alpha1` CRD that the operator watches.
4//! The spec mirrors the CLI manifest schema with additional fields for
5//! database connection and reconciliation scheduling.
6
7use kube::CustomResource;
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10use std::collections::{BTreeMap, BTreeSet};
11
12use pgroles_core::manifest::{
13    DefaultPrivilege, Grant, Membership, ObjectType, Privilege, RoleRetirement, SchemaBinding,
14};
15
16/// Valid PostgreSQL SSL modes for connection params.
17pub const VALID_SSL_MODES: &[&str] = &[
18    "disable",
19    "allow",
20    "prefer",
21    "require",
22    "verify-ca",
23    "verify-full",
24];
25
26// ---------------------------------------------------------------------------
27// CRD spec
28// ---------------------------------------------------------------------------
29
30/// Spec for a `PostgresPolicy` custom resource.
31///
32/// Defines the desired state of PostgreSQL roles, grants, default privileges,
33/// and memberships for a single database connection.
34#[derive(CustomResource, Debug, Clone, Serialize, Deserialize, JsonSchema)]
35#[kube(
36    group = "pgroles.io",
37    version = "v1alpha1",
38    kind = "PostgresPolicy",
39    namespaced,
40    status = "PostgresPolicyStatus",
41    shortname = "pgr",
42    category = "pgroles",
43    printcolumn = r#"{"name":"Ready","type":"string","jsonPath":".status.conditions[?(@.type==\"Ready\")].status"}"#,
44    printcolumn = r#"{"name":"Mode","type":"string","jsonPath":".spec.mode"}"#,
45    printcolumn = r#"{"name":"Recon","type":"string","jsonPath":".spec.reconciliation_mode","priority":1}"#,
46    printcolumn = r#"{"name":"Drift","type":"string","jsonPath":".status.conditions[?(@.type==\"Drifted\")].status"}"#,
47    printcolumn = r#"{"name":"Changes","type":"integer","jsonPath":".status.change_summary.total"}"#,
48    printcolumn = r#"{"name":"Last Reconcile","type":"date","jsonPath":".status.last_successful_reconcile_time"}"#,
49    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
50)]
51pub struct PostgresPolicySpec {
52    /// Database connection configuration.
53    pub connection: ConnectionSpec,
54
55    /// Reconciliation interval (e.g. "5m", "1h"). Defaults to "5m".
56    #[serde(default = "default_interval")]
57    pub interval: String,
58
59    /// Suspend reconciliation when true. Defaults to false.
60    #[serde(default)]
61    pub suspend: bool,
62
63    /// Reconciliation mode: `apply` executes SQL, `plan` computes drift only.
64    #[serde(default)]
65    pub mode: PolicyMode,
66
67    /// Convergence strategy: how aggressively to converge the database.
68    ///
69    /// - `authoritative` (default): full convergence — anything not in the
70    ///   manifest is revoked/dropped.
71    /// - `additive`: only grant, never revoke — safe for incremental adoption.
72    /// - `adopt`: manage declared roles fully, but never drop undeclared roles.
73    #[serde(default)]
74    pub reconciliation_mode: CrdReconciliationMode,
75
76    /// Default owner for ALTER DEFAULT PRIVILEGES (e.g. "app_owner").
77    #[serde(default)]
78    pub default_owner: Option<String>,
79
80    /// Reusable privilege profiles.
81    #[serde(default)]
82    pub profiles: std::collections::HashMap<String, ProfileSpec>,
83
84    /// Schema bindings that expand profiles into concrete roles/grants.
85    #[serde(default)]
86    pub schemas: Vec<SchemaBinding>,
87
88    /// One-off role definitions.
89    #[serde(default)]
90    pub roles: Vec<RoleSpec>,
91
92    /// One-off grants.
93    #[serde(default)]
94    pub grants: Vec<Grant>,
95
96    /// One-off default privileges.
97    #[serde(default)]
98    pub default_privileges: Vec<DefaultPrivilege>,
99
100    /// Membership edges.
101    #[serde(default)]
102    pub memberships: Vec<Membership>,
103
104    /// Explicit role-retirement workflows for roles that should be removed.
105    #[serde(default)]
106    pub retirements: Vec<RoleRetirement>,
107
108    /// Approval mode for plans: `auto` or `manual`.
109    /// When `manual`, plans require explicit approval before execution.
110    /// When `auto`, plans are approved and applied immediately.
111    /// When omitted, inferred from `mode`: `apply` → `auto`, `plan` → `manual`.
112    /// This ensures backward compatibility for existing `mode: apply` users.
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub approval: Option<ApprovalMode>,
115}
116
117fn default_interval() -> String {
118    "5m".to_string()
119}
120
121/// Policy reconcile mode.
122#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
123#[serde(rename_all = "lowercase")]
124pub enum PolicyMode {
125    #[default]
126    Apply,
127    Plan,
128}
129
130/// Convergence strategy for how aggressively to converge the database.
131#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
132#[serde(rename_all = "lowercase")]
133pub enum CrdReconciliationMode {
134    /// Full convergence — the manifest is the entire truth.
135    #[default]
136    Authoritative,
137    /// Only grant, never revoke — safe for incremental adoption.
138    Additive,
139    /// Manage declared roles fully, but never drop undeclared roles.
140    Adopt,
141}
142
143/// Approval mode for plans generated by this policy.
144#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
145pub enum ApprovalMode {
146    /// Plans require explicit approval annotation before execution.
147    #[serde(rename = "manual")]
148    Manual,
149    /// Plans are approved and applied automatically.
150    #[serde(rename = "auto")]
151    Auto,
152}
153
154impl PostgresPolicySpec {
155    /// Resolve the effective approval mode, inferring from `mode` when not set.
156    /// `apply` → `Auto` (backward compat), `plan` → `Manual`.
157    pub fn effective_approval(&self) -> ApprovalMode {
158        match &self.approval {
159            Some(mode) => mode.clone(),
160            None => match self.mode {
161                PolicyMode::Apply => ApprovalMode::Auto,
162                PolicyMode::Plan => ApprovalMode::Manual,
163            },
164        }
165    }
166}
167
168// ---------------------------------------------------------------------------
169// Well-known annotations and labels
170// ---------------------------------------------------------------------------
171
172/// Annotation key used to approve a `PostgresPolicyPlan`.
173pub const PLAN_APPROVED_ANNOTATION: &str = "pgroles.io/approved";
174
175/// Annotation key used to reject a `PostgresPolicyPlan`.
176pub const PLAN_REJECTED_ANNOTATION: &str = "pgroles.io/rejected";
177
178/// Annotation key used to request an immediate `PostgresPolicy` reconcile.
179pub const REQUESTED_RECONCILE_ANNOTATION: &str = "reconcile.pgroles.io/requestedAt";
180
181/// Label key for the parent policy name on plan resources.
182pub const LABEL_POLICY: &str = "pgroles.io/policy";
183
184/// Label key for the managed database identity on plan resources.
185pub const LABEL_DATABASE_IDENTITY: &str = "pgroles.io/database-identity";
186
187/// Label key for the plan name on SQL storage resources.
188pub const LABEL_PLAN: &str = "pgroles.io/plan";
189
190impl From<CrdReconciliationMode> for pgroles_core::diff::ReconciliationMode {
191    fn from(crd: CrdReconciliationMode) -> Self {
192        match crd {
193            CrdReconciliationMode::Authoritative => {
194                pgroles_core::diff::ReconciliationMode::Authoritative
195            }
196            CrdReconciliationMode::Additive => pgroles_core::diff::ReconciliationMode::Additive,
197            CrdReconciliationMode::Adopt => pgroles_core::diff::ReconciliationMode::Adopt,
198        }
199    }
200}
201
202/// Database connection configuration.
203///
204/// Supports two mutually exclusive modes:
205///
206/// **Mode 1 — Single URL** (backward-compatible):
207/// ```yaml
208/// connection:
209///   secretRef: { name: my-secret }
210///   secretKey: DATABASE_URL        # optional, defaults to DATABASE_URL
211/// ```
212///
213/// **Mode 2 — Structured params** (for Zalando/CNPG/PGO secrets):
214/// ```yaml
215/// connection:
216///   params:
217///     host: my-cluster-postgres
218///     port: 5432
219///     dbname: mydb
220///     usernameSecret: { name: zalando-creds, key: username }
221///     passwordSecret: { name: zalando-creds, key: password }
222/// ```
223///
224/// Params mode can also use provider-backed authentication instead of a static
225/// password, for example GKE Workload Identity to Cloud SQL IAM.
226#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
227#[serde(rename_all = "camelCase")]
228pub struct ConnectionSpec {
229    /// Reference to a Kubernetes Secret containing a connection URL.
230    /// Mutually exclusive with `params`.
231    #[serde(default)]
232    pub secret_ref: Option<SecretReference>,
233
234    /// Key within the Secret to read. Defaults to `DATABASE_URL`.
235    /// Only used with `secretRef`.
236    #[serde(default)]
237    pub secret_key: Option<String>,
238
239    /// Structured connection parameters. Each field is either a plain string
240    /// or a reference to a Secret key. Mutually exclusive with `secretRef`.
241    #[serde(default)]
242    pub params: Option<ConnectionParams>,
243}
244
245impl ConnectionSpec {
246    /// Effective secret key for URL mode. Defaults to `DATABASE_URL`.
247    pub fn effective_secret_key(&self) -> &str {
248        self.secret_key.as_deref().unwrap_or("DATABASE_URL")
249    }
250
251    /// Collect all Secret names referenced by this connection spec.
252    pub fn collect_secret_names(&self, names: &mut BTreeSet<String>) {
253        if let Some(ref secret_ref) = self.secret_ref {
254            names.insert(secret_ref.name.clone());
255        }
256        if let Some(ref params) = self.params {
257            for sel in [
258                &params.host_secret,
259                &params.port_secret,
260                &params.dbname_secret,
261                &params.username_secret,
262                &params.password_secret,
263                &params.ssl_mode_secret,
264            ]
265            .into_iter()
266            .flatten()
267            {
268                names.insert(sel.name.clone());
269            }
270        }
271    }
272
273    /// Deterministic identity key for this connection spec.
274    ///
275    /// - URL mode: `{secret_ref.name}/{secret_key}`
276    /// - Params mode: canonical representation of the params
277    ///
278    /// Uses `\0` as field separator since null bytes cannot appear in K8s names
279    /// or secret values, avoiding ambiguity from colons in literal values.
280    /// Deterministic identity key for per-database locking and conflict detection.
281    ///
282    /// Identifies the target database (host + port + dbname) but NOT the
283    /// credentials. Two policies targeting the same database with different
284    /// users should still be considered as targeting the same database for
285    /// locking and overlap checks.
286    pub fn identity_key(&self) -> String {
287        if let Some(ref secret_ref) = self.secret_ref {
288            format!("{}/{}", secret_ref.name, self.effective_secret_key())
289        } else if let Some(ref params) = self.params {
290            let port_part = params
291                .port
292                .as_ref()
293                .map(|p| format!("literal={p}"))
294                .or_else(|| {
295                    params
296                        .port_secret
297                        .as_ref()
298                        .map(|s| format!("secret={}\0{}", s.name, s.key))
299                })
300                .unwrap_or_else(|| "5432".to_string());
301            format!(
302                "params\0{}\0{}\0{}",
303                field_identity_repr(&params.host, &params.host_secret),
304                field_identity_repr(&params.dbname, &params.dbname_secret),
305                port_part,
306            )
307        } else {
308            "invalid-connection".to_string()
309        }
310    }
311
312    /// Cache key for pool lookup. Includes ALL connection params so that any
313    /// configuration change (credentials, sslMode, host, etc.) invalidates
314    /// the cached pool. This is strictly more specific than `identity_key`.
315    pub fn cache_key(&self, namespace: &str) -> String {
316        if let Some(ref params) = self.params {
317            let user_part = field_identity_repr(&params.username, &params.username_secret);
318            let pass_part = field_identity_repr(&params.password, &params.password_secret);
319            let auth_part = params
320                .auth
321                .as_ref()
322                .map(ConnectionAuth::cache_key)
323                .unwrap_or_default();
324            let ssl_part = params
325                .ssl_mode
326                .as_ref()
327                .map(|v| format!("literal={v}"))
328                .or_else(|| {
329                    params
330                        .ssl_mode_secret
331                        .as_ref()
332                        .map(|s| format!("secret={}\0{}", s.name, s.key))
333                })
334                .unwrap_or_default();
335            let role_part = params.set_role.as_deref().unwrap_or("");
336            format!(
337                "{namespace}/{}\0user={user_part}\0pass={pass_part}\0auth={auth_part}\0ssl={ssl_part}\0role={role_part}",
338                self.identity_key()
339            )
340        } else {
341            format!("{namespace}/{}", self.identity_key())
342        }
343    }
344}
345
346/// Deterministic string representation for a literal/secret field pair.
347///
348/// Uses a `literal=` / `secret=` prefix scheme so that a literal value
349/// can never collide with a secret reference representation.
350fn field_identity_repr(literal: &Option<String>, secret: &Option<SecretKeySelector>) -> String {
351    if let Some(value) = literal {
352        format!("literal={value}")
353    } else if let Some(sel) = secret {
354        format!("secret={}\0{}", sel.name, sel.key)
355    } else {
356        String::new()
357    }
358}
359
360/// Default OAuth scope used for Cloud SQL IAM database login tokens.
361pub const DEFAULT_GCP_CLOUD_SQL_LOGIN_SCOPE: &str =
362    "https://www.googleapis.com/auth/sqlservice.login";
363
364/// Structured connection parameters for building a PostgreSQL connection URL.
365///
366/// Each field supports either a literal value or a reference to a Kubernetes
367/// Secret key. For each parameter, set either the literal field or the
368/// corresponding `*Secret` field — not both.
369///
370/// ```yaml
371/// # Zalando pattern — literals for non-sensitive, secrets for credentials
372/// params:
373///   host: my-cluster-postgres
374///   port: 5432
375///   dbname: mydb
376///   sslMode: require
377///   usernameSecret:
378///     name: pg-creds
379///     key: username
380///   passwordSecret:
381///     name: pg-creds
382///     key: password
383/// ```
384#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
385#[serde(rename_all = "camelCase")]
386pub struct ConnectionParams {
387    /// PostgreSQL host as a literal value.
388    #[serde(default)]
389    pub host: Option<String>,
390    /// PostgreSQL host from a Secret key.
391    #[serde(default)]
392    pub host_secret: Option<SecretKeySelector>,
393
394    /// Port as a literal value. Defaults to 5432 if neither port nor portSecret is set.
395    #[serde(default)]
396    pub port: Option<u16>,
397    /// Port from a Secret key.
398    #[serde(default)]
399    pub port_secret: Option<SecretKeySelector>,
400
401    /// Database name as a literal value.
402    #[serde(default)]
403    pub dbname: Option<String>,
404    /// Database name from a Secret key.
405    #[serde(default)]
406    pub dbname_secret: Option<SecretKeySelector>,
407
408    /// Username as a literal value.
409    #[serde(default)]
410    pub username: Option<String>,
411    /// Username from a Secret key.
412    #[serde(default)]
413    pub username_secret: Option<SecretKeySelector>,
414
415    /// Password as a literal value (not recommended for production).
416    #[serde(default)]
417    pub password: Option<String>,
418    /// Password from a Secret key (recommended).
419    #[serde(default)]
420    pub password_secret: Option<SecretKeySelector>,
421
422    /// Provider-backed authentication for connections that use short-lived
423    /// credentials instead of a static PostgreSQL password.
424    #[serde(default)]
425    pub auth: Option<ConnectionAuth>,
426
427    /// SSL mode as a literal value.
428    #[serde(default)]
429    pub ssl_mode: Option<String>,
430    /// SSL mode from a Secret key.
431    #[serde(default)]
432    pub ssl_mode_secret: Option<SecretKeySelector>,
433
434    /// Run `SET ROLE "<value>"` once on every pooled connection. Useful when
435    /// the operator authenticates as a low-privilege identity (e.g. a Cloud
436    /// SQL IAM user) that has been granted membership in a privileged role
437    /// like `cloudsqlsuperuser` — PostgreSQL does not inherit role
438    /// *attributes* (`CREATEROLE`, `CREATEDB`, …) through `GRANT … TO …`, so
439    /// `SET ROLE` is required for the connection to act with the parent
440    /// role's attributes.
441    ///
442    /// Must be a simple PostgreSQL identifier matching
443    /// `^[A-Za-z_][A-Za-z0-9_$-]*$`. The pattern intentionally excludes `@`
444    /// and `.` — `setRole` is for switching to a privileged *group* role
445    /// (e.g. `cloudsqlsuperuser`), not an IAM-style principal like
446    /// `pgroles-operator@project.iam`, which has no extra attributes to
447    /// inherit via `SET ROLE`.
448    #[serde(default)]
449    #[schemars(regex(pattern = SET_ROLE_PATTERN))]
450    pub set_role: Option<String>,
451}
452
453/// Provider-backed authentication for `connection.params`.
454#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
455#[serde(tag = "type")]
456pub enum ConnectionAuth {
457    /// Fetch a Cloud SQL IAM database login token using the GKE metadata
458    /// server. `username` must be the Cloud SQL PostgreSQL IAM role name
459    /// (for service accounts, the email without `.gserviceaccount.com`).
460    #[serde(rename = "gcp_workload_identity", rename_all = "camelCase")]
461    GcpWorkloadIdentity {
462        /// Target Google service account to impersonate before requesting the
463        /// Cloud SQL login token. Omit to use the pod's bound identity.
464        #[serde(default)]
465        impersonate_service_account: Option<String>,
466        /// OAuth scope requested for the access token.
467        #[serde(default)]
468        scope: Option<String>,
469    },
470}
471
472impl ConnectionAuth {
473    pub fn gcp_scope(&self) -> &str {
474        match self {
475            Self::GcpWorkloadIdentity { scope, .. } => scope
476                .as_deref()
477                .unwrap_or(DEFAULT_GCP_CLOUD_SQL_LOGIN_SCOPE),
478        }
479    }
480
481    pub fn gcp_impersonate_service_account(&self) -> Option<&str> {
482        match self {
483            Self::GcpWorkloadIdentity {
484                impersonate_service_account,
485                ..
486            } => impersonate_service_account.as_deref(),
487        }
488    }
489
490    fn cache_key(&self) -> String {
491        match self {
492            Self::GcpWorkloadIdentity {
493                impersonate_service_account,
494                scope,
495            } => format!(
496                "gcp_workload_identity\0impersonate={}\0scope={}",
497                impersonate_service_account.as_deref().unwrap_or_default(),
498                scope
499                    .as_deref()
500                    .unwrap_or(DEFAULT_GCP_CLOUD_SQL_LOGIN_SCOPE)
501            ),
502        }
503    }
504}
505
506/// Reference to a specific key within a Kubernetes Secret.
507#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
508pub struct SecretKeySelector {
509    /// Name of the Secret.
510    pub name: String,
511    /// Key within the Secret.
512    pub key: String,
513}
514
515/// Reference to a Kubernetes Secret in the same namespace.
516#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
517pub struct SecretReference {
518    /// Name of the Secret.
519    pub name: String,
520}
521
522/// A reusable privilege profile (CRD-compatible version).
523///
524/// This mirrors `pgroles_core::manifest::Profile` but derives `JsonSchema`.
525#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
526pub struct ProfileSpec {
527    #[serde(default)]
528    pub login: Option<bool>,
529
530    #[serde(default)]
531    pub inherit: Option<bool>,
532
533    #[serde(default)]
534    pub grants: Vec<ProfileGrantSpec>,
535
536    #[serde(default)]
537    pub default_privileges: Vec<DefaultPrivilegeGrantSpec>,
538}
539
540/// Grant template within a profile.
541#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
542pub struct ProfileGrantSpec {
543    pub privileges: Vec<Privilege>,
544    #[serde(alias = "on")]
545    pub object: ProfileObjectTargetSpec,
546}
547
548/// Object target within a profile.
549#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
550pub struct ProfileObjectTargetSpec {
551    #[serde(rename = "type")]
552    pub object_type: ObjectType,
553    #[serde(default)]
554    pub name: Option<String>,
555}
556
557/// Default privilege grant within a profile.
558#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
559pub struct DefaultPrivilegeGrantSpec {
560    #[serde(default)]
561    pub role: Option<String>,
562    pub privileges: Vec<Privilege>,
563    pub on_type: ObjectType,
564}
565
566/// A concrete role definition (CRD-compatible version).
567#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
568pub struct RoleSpec {
569    pub name: String,
570    /// Treat this role as externally managed. The operator may reference it in
571    /// grants, ownership, and memberships, but will not create, alter, drop,
572    /// password-manage, or manage memberships granted from this role.
573    #[serde(default)]
574    pub external: bool,
575    #[serde(default)]
576    pub login: Option<bool>,
577    #[serde(default)]
578    pub superuser: Option<bool>,
579    #[serde(default)]
580    pub createdb: Option<bool>,
581    #[serde(default)]
582    pub createrole: Option<bool>,
583    #[serde(default)]
584    pub inherit: Option<bool>,
585    #[serde(default)]
586    pub replication: Option<bool>,
587    #[serde(default)]
588    pub bypassrls: Option<bool>,
589    #[serde(default)]
590    pub connection_limit: Option<i32>,
591    #[serde(default)]
592    pub comment: Option<String>,
593    /// Password source for this role. Either a reference to an existing Secret
594    /// or a request for the operator to generate one.
595    #[serde(default)]
596    pub password: Option<PasswordSpec>,
597    /// Password expiration timestamp (ISO 8601, e.g. "2025-12-31T00:00:00Z").
598    #[serde(default)]
599    pub password_valid_until: Option<String>,
600}
601
602/// Password configuration: either reference an existing Secret or have the
603/// operator generate a password and create a Secret.
604///
605/// Exactly one of `secretRef` or `generate` must be set.
606///
607/// ```yaml
608/// # Read from existing Secret:
609/// password:
610///   secretRef: { name: role-passwords }
611///   secretKey: password-user
612///
613/// # Operator generates and manages a Secret:
614/// password:
615///   generate:
616///     length: 48
617/// ```
618#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
619#[serde(rename_all = "camelCase")]
620pub struct PasswordSpec {
621    /// Reference to an existing Kubernetes Secret containing the password.
622    /// Mutually exclusive with `generate`.
623    #[serde(default)]
624    pub secret_ref: Option<SecretReference>,
625    /// Key within the referenced Secret. Defaults to the role name.
626    /// Only used with `secretRef`.
627    #[serde(default)]
628    pub secret_key: Option<String>,
629    /// Generate a random password and store it in a new Kubernetes Secret.
630    /// Mutually exclusive with `secretRef`.
631    #[serde(default)]
632    pub generate: Option<GeneratePasswordSpec>,
633}
634
635impl PasswordSpec {
636    /// Returns true if this is a reference to an existing Secret.
637    pub fn is_secret_ref(&self) -> bool {
638        self.secret_ref.is_some()
639    }
640
641    /// Returns true if this is a request to generate a password.
642    pub fn is_generate(&self) -> bool {
643        self.generate.is_some()
644    }
645}
646
647/// Configuration for operator-generated passwords.
648#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
649#[serde(rename_all = "camelCase")]
650pub struct GeneratePasswordSpec {
651    /// Password length. Defaults to 32. Minimum 16, maximum 128.
652    #[serde(default)]
653    pub length: Option<u32>,
654    /// Override the generated Secret name. Defaults to `{policy}-pgr-{role}`.
655    #[serde(default)]
656    pub secret_name: Option<String>,
657    /// Key within the generated Secret. Defaults to `password`.
658    #[serde(default)]
659    pub secret_key: Option<String>,
660}
661
662#[derive(Debug, Clone, thiserror::Error)]
663pub enum PasswordValidationError {
664    #[error("role \"{role}\" has a password but login is not enabled")]
665    PasswordWithoutLogin { role: String },
666
667    #[error("role \"{role}\" password must set exactly one of secretRef or generate")]
668    InvalidPasswordMode { role: String },
669
670    #[error("role \"{role}\" password.generate.length must be between {min} and {max}")]
671    InvalidGeneratedLength { role: String, min: u32, max: u32 },
672
673    #[error(
674        "role \"{role}\" password.generate.secretName \"{name}\" is not a valid Kubernetes Secret name"
675    )]
676    InvalidGeneratedSecretName { role: String, name: String },
677
678    #[error("role \"{role}\" password {field} \"{key}\" is not a valid Kubernetes Secret data key")]
679    InvalidSecretKey {
680        role: String,
681        field: &'static str,
682        key: String,
683    },
684
685    #[error(
686        "role \"{role}\" password.generate.secretKey \"{key}\" is reserved for the SCRAM verifier"
687    )]
688    ReservedGeneratedSecretKey { role: String, key: String },
689}
690
691/// Errors from connection spec validation.
692#[derive(Debug, Clone, thiserror::Error)]
693pub enum ConnectionValidationError {
694    #[error("connection: exactly one of secretRef or params must be set, but both were provided")]
695    BothModesSet,
696
697    #[error("connection: exactly one of secretRef or params must be set, but neither was provided")]
698    NeitherModeSet,
699
700    #[error("connection.params.{field}: secret {detail}")]
701    EmptySecretKeyRef { field: String, detail: String },
702
703    #[error(
704        "connection.params.sslMode: \"{value}\" is not valid (expected one of: disable, allow, prefer, require, verify-ca, verify-full)"
705    )]
706    InvalidSslMode { value: String },
707
708    #[error("connection.params.{field}: literal value must not be empty or whitespace-only")]
709    EmptyLiteral { field: String },
710
711    #[error("connection.params: exactly one of {field} or {field}Secret must be set")]
712    NeitherFieldSet { field: String },
713
714    #[error(
715        "connection.params: only one of {field} or {field}Secret may be set, but both were provided"
716    )]
717    BothFieldsSet { field: String },
718
719    #[error("connection.params.auth: {field} must not be empty or whitespace-only")]
720    EmptyAuthField { field: String },
721
722    #[error("connection.params: password/passwordSecret are mutually exclusive with auth")]
723    AuthWithPassword,
724
725    #[error(
726        "connection.params.setRole: \"{value}\" is not a valid PostgreSQL role identifier (must match {pattern})",
727        pattern = SET_ROLE_PATTERN,
728    )]
729    InvalidRoleName { value: String },
730}
731
732/// Regex pattern restricting `connection.params.setRole` values.
733///
734/// Single source of truth: used by the `#[schemars(...)]` attribute on the
735/// field (emitted into the CRD's OpenAPI schema), referenced by the
736/// `InvalidRoleName` error message, and pinned by `is_valid_set_role_identifier`
737/// via unit tests.
738///
739/// Intentionally rejects `@` and `.`, so IAM-email-style identifiers
740/// (e.g. `pgroles-operator@project.iam`) cannot be a `setRole` target.
741/// `SET ROLE` is meant for switching to a privileged *group* role like
742/// `cloudsqlsuperuser`; IAM principals don't carry role attributes worth
743/// switching to.
744pub(crate) const SET_ROLE_PATTERN: &str = "^[A-Za-z_][A-Za-z0-9_$-]*$";
745
746/// Returns true if `s` is a simple PostgreSQL role identifier matching
747/// [`SET_ROLE_PATTERN`].
748///
749/// `SET ROLE` does not accept bind parameters, so any value reaching the
750/// connection-pool hook is interpolated into the SQL string. Restricting
751/// identifiers at admission time is defence in depth on top of the
752/// double-quoting in the `after_connect` callback.
753pub(crate) fn is_valid_set_role_identifier(s: &str) -> bool {
754    let mut bytes = s.bytes();
755    match bytes.next() {
756        Some(b) if b.is_ascii_alphabetic() || b == b'_' => {}
757        _ => return false,
758    }
759    bytes.all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'$' || b == b'-')
760}
761
762/// Validate a Kubernetes Secret name per RFC 1123 DNS subdomain rules:
763/// lowercase alpha start, alphanumeric end, body allows lowercase alpha,
764/// digits, `-`, and `.`.
765fn is_valid_secret_name(name: &str) -> bool {
766    if name.is_empty() || name.len() > crate::password::MAX_SECRET_NAME_LENGTH {
767        return false;
768    }
769    let bytes = name.as_bytes();
770    // RFC 1123: must start with a lowercase letter.
771    if !bytes[0].is_ascii_lowercase() {
772        return false;
773    }
774    if !bytes[bytes.len() - 1].is_ascii_lowercase() && !bytes[bytes.len() - 1].is_ascii_digit() {
775        return false;
776    }
777    bytes
778        .iter()
779        .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || *b == b'-' || *b == b'.')
780}
781
782fn is_valid_secret_key(key: &str) -> bool {
783    !key.is_empty()
784        && key
785            .bytes()
786            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))
787}
788
789// ---------------------------------------------------------------------------
790// CRD status
791// ---------------------------------------------------------------------------
792
793/// Status of a `PostgresPolicy` resource.
794#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
795pub struct PostgresPolicyStatus {
796    /// Standard Kubernetes conditions.
797    #[serde(default)]
798    pub conditions: Vec<PolicyCondition>,
799
800    /// The `.metadata.generation` that was last successfully reconciled.
801    #[serde(default)]
802    pub observed_generation: Option<i64>,
803
804    /// The `.metadata.generation` that was last attempted.
805    #[serde(default)]
806    pub last_attempted_generation: Option<i64>,
807
808    /// ISO 8601 timestamp of the last successful reconciliation.
809    #[serde(default)]
810    pub last_successful_reconcile_time: Option<String>,
811
812    /// Deprecated alias retained for compatibility with older status readers.
813    #[serde(default)]
814    pub last_reconcile_time: Option<String>,
815
816    /// Last force-reconcile annotation value handled by the operator.
817    #[serde(default, rename = "lastHandledReconcileAt")]
818    pub last_handled_reconcile_at: Option<String>,
819
820    /// Summary of changes applied in the last reconciliation.
821    #[serde(default)]
822    pub change_summary: Option<ChangeSummary>,
823
824    /// The reconciliation mode used for the last successful reconcile.
825    #[serde(default)]
826    pub last_reconcile_mode: Option<PolicyMode>,
827
828    /// Planned SQL for the last successful plan-mode reconcile.
829    #[serde(default)]
830    pub planned_sql: Option<String>,
831
832    /// Whether `planned_sql` was truncated to fit safely in status.
833    #[serde(default)]
834    pub planned_sql_truncated: bool,
835
836    /// Canonical identity of the managed database target.
837    #[serde(default)]
838    pub managed_database_identity: Option<String>,
839
840    /// Roles claimed by this policy's declared ownership scope.
841    #[serde(default)]
842    pub owned_roles: Vec<String>,
843
844    /// Schemas claimed by this policy's declared ownership scope.
845    #[serde(default)]
846    pub owned_schemas: Vec<String>,
847
848    /// Last reconcile error message, if any.
849    #[serde(default)]
850    pub last_error: Option<String>,
851
852    /// Last applied password source version for each password-managed role.
853    #[serde(default)]
854    pub applied_password_source_versions: BTreeMap<String, String>,
855
856    /// Consecutive transient operational failures used for exponential backoff.
857    #[serde(default)]
858    pub transient_failure_count: i32,
859
860    /// Reference to the current/latest plan for this policy.
861    #[serde(default)]
862    pub current_plan_ref: Option<PlanReference>,
863}
864
865/// A condition on the `PostgresPolicy` resource.
866#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
867pub struct PolicyCondition {
868    /// Type of condition: "Ready", "Reconciling", "Degraded".
869    #[serde(rename = "type")]
870    pub condition_type: String,
871
872    /// Status: "True", "False", or "Unknown".
873    pub status: String,
874
875    /// Human-readable reason for the condition.
876    #[serde(default)]
877    pub reason: Option<String>,
878
879    /// Human-readable message.
880    #[serde(default)]
881    pub message: Option<String>,
882
883    /// Last time the condition transitioned.
884    #[serde(default)]
885    pub last_transition_time: Option<String>,
886}
887
888/// Reference to a `PostgresPolicyPlan` resource.
889#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
890pub struct PlanReference {
891    pub name: String,
892}
893
894/// Summary of changes applied during reconciliation.
895#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
896#[serde(default)]
897pub struct ChangeSummary {
898    #[serde(default)]
899    pub roles_created: i32,
900    #[serde(default)]
901    pub roles_altered: i32,
902    #[serde(default)]
903    pub schemas_created: i32,
904    #[serde(default)]
905    pub schema_owners_altered: i32,
906    #[serde(default)]
907    pub roles_dropped: i32,
908    #[serde(default)]
909    pub sessions_terminated: i32,
910    #[serde(default)]
911    pub grants_added: i32,
912    #[serde(default)]
913    pub grants_revoked: i32,
914    #[serde(default)]
915    pub default_privileges_set: i32,
916    #[serde(default)]
917    pub default_privileges_revoked: i32,
918    #[serde(default)]
919    pub members_added: i32,
920    #[serde(default)]
921    pub members_removed: i32,
922    #[serde(default)]
923    pub passwords_set: i32,
924    #[serde(default)]
925    pub total: i32,
926}
927
928// ---------------------------------------------------------------------------
929// PostgresPolicyPlan CRD
930// ---------------------------------------------------------------------------
931
932/// Spec for a `PostgresPolicyPlan` custom resource.
933///
934/// Represents a computed reconciliation plan for a `PostgresPolicy`. Plans are
935/// created by the operator and may require explicit approval before execution.
936#[derive(CustomResource, Debug, Clone, Serialize, Deserialize, JsonSchema)]
937#[kube(
938    group = "pgroles.io",
939    version = "v1alpha1",
940    kind = "PostgresPolicyPlan",
941    namespaced,
942    status = "PostgresPolicyPlanStatus",
943    shortname = "pgplan",
944    category = "pgroles",
945    printcolumn = r#"{"name":"Policy","type":"string","jsonPath":".spec.policyRef.name"}"#,
946    printcolumn = r#"{"name":"Mode","type":"string","jsonPath":".spec.reconciliationMode"}"#,
947    printcolumn = r#"{"name":"Approved","type":"string","jsonPath":".status.conditions[?(@.type==\"Approved\")].status"}"#,
948    printcolumn = r#"{"name":"Changes","type":"integer","jsonPath":".status.changeSummary.total"}"#,
949    printcolumn = r#"{"name":"SQL Stmts","type":"integer","jsonPath":".status.sqlStatements","priority":1}"#,
950    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
951    printcolumn = r#"{"name":"SQL","type":"string","jsonPath":".status.sqlRef.name","priority":1}"#,
952    printcolumn = r#"{"name":"Hash","type":"string","jsonPath":".status.sqlHash","priority":1}"#,
953    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
954)]
955#[serde(rename_all = "camelCase")]
956pub struct PostgresPolicyPlanSpec {
957    /// Reference to the policy that generated this plan.
958    pub policy_ref: PolicyPlanRef,
959    /// The policy's `.metadata.generation` at plan time.
960    pub policy_generation: i64,
961    /// Reconciliation mode used for this plan.
962    pub reconciliation_mode: CrdReconciliationMode,
963    /// Roles that this plan covers.
964    #[serde(default)]
965    pub owned_roles: Vec<String>,
966    /// Schemas that this plan covers.
967    #[serde(default)]
968    pub owned_schemas: Vec<String>,
969    /// Database identity string for disambiguation in multi-db setups.
970    pub managed_database_identity: String,
971}
972
973/// Reference to the parent `PostgresPolicy` that generated a plan.
974#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
975pub struct PolicyPlanRef {
976    pub name: String,
977}
978
979/// Status of a `PostgresPolicyPlan` resource.
980#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
981#[serde(rename_all = "camelCase")]
982pub struct PostgresPolicyPlanStatus {
983    /// Phase: Pending, Approved, Applying, Applied, Failed, Superseded.
984    #[serde(default)]
985    pub phase: PlanPhase,
986    /// Standard conditions: Computed, Approved, Applied.
987    #[serde(default)]
988    pub conditions: Vec<PolicyCondition>,
989    /// Summary of changes in this plan.
990    #[serde(default)]
991    pub change_summary: Option<ChangeSummary>,
992    /// Reference to ConfigMap containing the full SQL (for large plans).
993    #[serde(default)]
994    pub sql_ref: Option<SqlRef>,
995    /// Inline SQL for small plans (below a size threshold).
996    #[serde(default)]
997    pub sql_inline: Option<String>,
998    /// True when the SQL preview was truncated because the full redacted SQL
999    /// could not be persisted within Kubernetes object limits.
1000    #[serde(default)]
1001    pub sql_truncated: bool,
1002    /// Timestamp when the plan was computed.
1003    #[serde(default)]
1004    pub computed_at: Option<String>,
1005    /// Timestamp when the plan was applied (if applicable).
1006    #[serde(default)]
1007    pub applied_at: Option<String>,
1008    /// Error message if apply failed.
1009    #[serde(default)]
1010    pub last_error: Option<String>,
1011    /// SHA-256 hash of the planned SQL, used to detect duplicate plans.
1012    /// If a newly computed plan has the same hash as the current pending plan,
1013    /// the operator can skip creating a redundant plan.
1014    #[serde(default)]
1015    pub sql_hash: Option<String>,
1016    /// Timestamp when the plan entered Applying phase (for stuck detection).
1017    #[serde(default)]
1018    pub applying_since: Option<String>,
1019    /// Timestamp when the plan entered Failed phase (for dedup window).
1020    #[serde(default)]
1021    pub failed_at: Option<String>,
1022    /// Number of SQL statements in the plan (after wildcard expansion).
1023    /// May be significantly larger than `changeSummary.total` when wildcard
1024    /// grants expand to many per-object statements.
1025    #[serde(default)]
1026    pub sql_statements: Option<i64>,
1027    /// SHA-256 hash of the redacted SQL preview bytes. This is for storage
1028    /// integrity only; approval and deduplication continue to use `sql_hash`.
1029    #[serde(default)]
1030    pub redacted_sql_hash: Option<String>,
1031    /// Uncompressed byte length of the redacted SQL preview.
1032    #[serde(default)]
1033    pub sql_original_bytes: Option<i64>,
1034    /// Stored byte length of the SQL preview after inline/truncation/compression.
1035    #[serde(default)]
1036    pub sql_stored_bytes: Option<i64>,
1037}
1038
1039/// Reference to a ConfigMap containing SQL for a plan.
1040#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
1041pub struct SqlRef {
1042    pub name: String,
1043    pub key: String,
1044    /// Compression used for the referenced SQL content. Missing means older
1045    /// uncompressed ConfigMap data.
1046    #[serde(default, skip_serializing_if = "Option::is_none")]
1047    pub compression: Option<SqlCompression>,
1048}
1049
1050/// Compression format used for persisted plan SQL previews.
1051#[derive(Debug, Clone, Copy, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
1052#[serde(rename_all = "lowercase")]
1053pub enum SqlCompression {
1054    Gzip,
1055}
1056
1057/// Phase of a `PostgresPolicyPlan`.
1058#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
1059pub enum PlanPhase {
1060    #[default]
1061    Pending,
1062    Approved,
1063    Applying,
1064    Applied,
1065    Failed,
1066    Superseded,
1067    Rejected,
1068}
1069
1070impl std::fmt::Display for PlanPhase {
1071    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1072        match self {
1073            PlanPhase::Pending => write!(f, "Pending"),
1074            PlanPhase::Approved => write!(f, "Approved"),
1075            PlanPhase::Applying => write!(f, "Applying"),
1076            PlanPhase::Applied => write!(f, "Applied"),
1077            PlanPhase::Failed => write!(f, "Failed"),
1078            PlanPhase::Superseded => write!(f, "Superseded"),
1079            PlanPhase::Rejected => write!(f, "Rejected"),
1080        }
1081    }
1082}
1083
1084// ---------------------------------------------------------------------------
1085// Conflict detection
1086// ---------------------------------------------------------------------------
1087
1088/// Canonical target identity for conflict detection between policies.
1089#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
1090pub struct DatabaseIdentity(String);
1091
1092impl DatabaseIdentity {
1093    /// Create a database identity from the namespace and connection spec's identity key.
1094    pub fn from_connection(namespace: &str, connection: &ConnectionSpec) -> Self {
1095        Self(format!("{namespace}/{}", connection.identity_key()))
1096    }
1097
1098    pub fn as_str(&self) -> &str {
1099        &self.0
1100    }
1101}
1102
1103/// Conservative ownership claims for a policy.
1104#[derive(Debug, Clone, Default, PartialEq, Eq)]
1105pub struct OwnershipClaims {
1106    pub roles: BTreeSet<String>,
1107    pub schemas: BTreeSet<String>,
1108}
1109
1110impl OwnershipClaims {
1111    pub fn overlaps(&self, other: &Self) -> bool {
1112        !self.roles.is_disjoint(&other.roles) || !self.schemas.is_disjoint(&other.schemas)
1113    }
1114
1115    pub fn overlap_summary(&self, other: &Self) -> String {
1116        let overlapping_roles: Vec<_> = self.roles.intersection(&other.roles).cloned().collect();
1117        let overlapping_schemas: Vec<_> =
1118            self.schemas.intersection(&other.schemas).cloned().collect();
1119
1120        let mut parts = Vec::new();
1121        if !overlapping_roles.is_empty() {
1122            parts.push(format!("roles: {}", overlapping_roles.join(", ")));
1123        }
1124        if !overlapping_schemas.is_empty() {
1125            parts.push(format!("schemas: {}", overlapping_schemas.join(", ")));
1126        }
1127
1128        parts.join("; ")
1129    }
1130}
1131
1132// ---------------------------------------------------------------------------
1133// Secret name helpers
1134// ---------------------------------------------------------------------------
1135
1136impl PostgresPolicySpec {
1137    pub fn validate_password_specs(
1138        &self,
1139        policy_name: &str,
1140    ) -> Result<(), PasswordValidationError> {
1141        for role in &self.roles {
1142            let Some(password) = &role.password else {
1143                continue;
1144            };
1145
1146            if role.login != Some(true) {
1147                return Err(PasswordValidationError::PasswordWithoutLogin {
1148                    role: role.name.clone(),
1149                });
1150            }
1151
1152            match (&password.secret_ref, &password.generate) {
1153                (Some(_), None) => {
1154                    let secret_key = password.secret_key.as_deref().unwrap_or(&role.name);
1155                    if !is_valid_secret_key(secret_key) {
1156                        return Err(PasswordValidationError::InvalidSecretKey {
1157                            role: role.name.clone(),
1158                            field: "secretKey",
1159                            key: secret_key.to_string(),
1160                        });
1161                    }
1162                }
1163                (None, Some(generate)) => {
1164                    if let Some(length) = generate.length
1165                        && !(crate::password::MIN_PASSWORD_LENGTH
1166                            ..=crate::password::MAX_PASSWORD_LENGTH)
1167                            .contains(&length)
1168                    {
1169                        return Err(PasswordValidationError::InvalidGeneratedLength {
1170                            role: role.name.clone(),
1171                            min: crate::password::MIN_PASSWORD_LENGTH,
1172                            max: crate::password::MAX_PASSWORD_LENGTH,
1173                        });
1174                    }
1175
1176                    let secret_name =
1177                        crate::password::generated_secret_name(policy_name, &role.name, generate);
1178                    if !is_valid_secret_name(&secret_name) {
1179                        return Err(PasswordValidationError::InvalidGeneratedSecretName {
1180                            role: role.name.clone(),
1181                            name: secret_name,
1182                        });
1183                    }
1184
1185                    let secret_key = crate::password::generated_secret_key(generate);
1186                    if !is_valid_secret_key(&secret_key) {
1187                        return Err(PasswordValidationError::InvalidSecretKey {
1188                            role: role.name.clone(),
1189                            field: "generate.secretKey",
1190                            key: secret_key,
1191                        });
1192                    }
1193                    if secret_key == crate::password::GENERATED_VERIFIER_KEY {
1194                        return Err(PasswordValidationError::ReservedGeneratedSecretKey {
1195                            role: role.name.clone(),
1196                            key: secret_key,
1197                        });
1198                    }
1199                }
1200                _ => {
1201                    return Err(PasswordValidationError::InvalidPasswordMode {
1202                        role: role.name.clone(),
1203                    });
1204                }
1205            }
1206        }
1207
1208        Ok(())
1209    }
1210
1211    /// Validate the connection spec.
1212    ///
1213    /// Ensures exactly one of `secretRef` or `params` is set, and that params
1214    /// mode has all required fields with valid values.
1215    pub fn validate_connection_spec(&self) -> Result<(), ConnectionValidationError> {
1216        let conn = &self.connection;
1217        match (&conn.secret_ref, &conn.params) {
1218            (Some(_), None) => {
1219                // URL mode — valid.
1220                Ok(())
1221            }
1222            (None, Some(params)) => {
1223                // Validate a required field pair: exactly one must be set.
1224                fn validate_required_field(
1225                    field: &str,
1226                    literal: &Option<String>,
1227                    secret: &Option<SecretKeySelector>,
1228                ) -> Result<(), ConnectionValidationError> {
1229                    match (literal, secret) {
1230                        (Some(_), Some(_)) => {
1231                            return Err(ConnectionValidationError::BothFieldsSet {
1232                                field: field.to_string(),
1233                            });
1234                        }
1235                        (None, None) => {
1236                            return Err(ConnectionValidationError::NeitherFieldSet {
1237                                field: field.to_string(),
1238                            });
1239                        }
1240                        (Some(s), None) => {
1241                            if s.trim().is_empty() {
1242                                return Err(ConnectionValidationError::EmptyLiteral {
1243                                    field: field.to_string(),
1244                                });
1245                            }
1246                        }
1247                        (None, Some(sel)) => {
1248                            validate_secret_selector(field, sel)?;
1249                        }
1250                    }
1251                    Ok(())
1252                }
1253
1254                // Validate an optional field pair: at most one may be set.
1255                fn validate_optional_field(
1256                    field: &str,
1257                    literal: &Option<impl AsRef<str>>,
1258                    secret: &Option<SecretKeySelector>,
1259                ) -> Result<(), ConnectionValidationError> {
1260                    let has_literal = literal.is_some();
1261                    if has_literal && secret.is_some() {
1262                        return Err(ConnectionValidationError::BothFieldsSet {
1263                            field: field.to_string(),
1264                        });
1265                    }
1266                    if let Some(s) = literal
1267                        && s.as_ref().trim().is_empty()
1268                    {
1269                        return Err(ConnectionValidationError::EmptyLiteral {
1270                            field: field.to_string(),
1271                        });
1272                    }
1273                    if let Some(sel) = secret {
1274                        validate_secret_selector(field, sel)?;
1275                    }
1276                    Ok(())
1277                }
1278
1279                fn validate_secret_selector(
1280                    field: &str,
1281                    sel: &SecretKeySelector,
1282                ) -> Result<(), ConnectionValidationError> {
1283                    if sel.name.trim().is_empty() {
1284                        return Err(ConnectionValidationError::EmptySecretKeyRef {
1285                            field: field.to_string(),
1286                            detail: "name must not be empty".to_string(),
1287                        });
1288                    }
1289                    if sel.key.trim().is_empty() {
1290                        return Err(ConnectionValidationError::EmptySecretKeyRef {
1291                            field: field.to_string(),
1292                            detail: "key must not be empty".to_string(),
1293                        });
1294                    }
1295                    Ok(())
1296                }
1297
1298                // Required fields: host, dbname, username. Password is only
1299                // required for static-password auth.
1300                validate_required_field("host", &params.host, &params.host_secret)?;
1301                validate_required_field("dbname", &params.dbname, &params.dbname_secret)?;
1302                validate_required_field("username", &params.username, &params.username_secret)?;
1303                if let Some(auth) = &params.auth {
1304                    if params.password.is_some() || params.password_secret.is_some() {
1305                        return Err(ConnectionValidationError::AuthWithPassword);
1306                    }
1307                    match auth {
1308                        ConnectionAuth::GcpWorkloadIdentity {
1309                            impersonate_service_account,
1310                            scope,
1311                        } => {
1312                            if let Some(value) = impersonate_service_account
1313                                && value.trim().is_empty()
1314                            {
1315                                return Err(ConnectionValidationError::EmptyAuthField {
1316                                    field: "impersonateServiceAccount".to_string(),
1317                                });
1318                            }
1319                            if let Some(value) = scope
1320                                && value.trim().is_empty()
1321                            {
1322                                return Err(ConnectionValidationError::EmptyAuthField {
1323                                    field: "scope".to_string(),
1324                                });
1325                            }
1326                        }
1327                    }
1328                } else {
1329                    validate_required_field("password", &params.password, &params.password_secret)?;
1330                }
1331
1332                // Optional fields: port, sslMode.
1333                // Port is u16 so we wrap it for the generic check.
1334                let port_str = params.port.map(|p| p.to_string());
1335                validate_optional_field("port", &port_str, &params.port_secret)?;
1336
1337                validate_optional_field("sslMode", &params.ssl_mode, &params.ssl_mode_secret)?;
1338
1339                // Validate sslMode value if it's a literal.
1340                if let Some(value) = &params.ssl_mode
1341                    && !VALID_SSL_MODES.contains(&value.as_str())
1342                {
1343                    return Err(ConnectionValidationError::InvalidSslMode {
1344                        value: value.clone(),
1345                    });
1346                }
1347
1348                // Validate setRole identifier. `SET ROLE` does not accept bind
1349                // params, so the identifier is restricted at admission time.
1350                if let Some(value) = &params.set_role {
1351                    if value.trim().is_empty() {
1352                        return Err(ConnectionValidationError::EmptyLiteral {
1353                            field: "setRole".to_string(),
1354                        });
1355                    }
1356                    if !is_valid_set_role_identifier(value) {
1357                        return Err(ConnectionValidationError::InvalidRoleName {
1358                            value: value.clone(),
1359                        });
1360                    }
1361                }
1362
1363                Ok(())
1364            }
1365            (Some(_), Some(_)) => Err(ConnectionValidationError::BothModesSet),
1366            (None, None) => Err(ConnectionValidationError::NeitherModeSet),
1367        }
1368    }
1369
1370    /// All Kubernetes Secret names referenced by this spec.
1371    ///
1372    /// Includes the connection Secret, password `secretRef` Secrets, and
1373    /// generated password Secrets. Used by the controller to trigger
1374    /// reconciliation when any of these Secrets change (or are deleted).
1375    pub fn referenced_secret_names(&self, policy_name: &str) -> BTreeSet<String> {
1376        let mut names = BTreeSet::new();
1377        // Connection secrets — either URL mode or structured params.
1378        self.connection.collect_secret_names(&mut names);
1379        for role in &self.roles {
1380            if let Some(pw) = &role.password {
1381                if let Some(secret_ref) = &pw.secret_ref {
1382                    names.insert(secret_ref.name.clone());
1383                }
1384                if let Some(gen_spec) = &pw.generate {
1385                    let secret_name =
1386                        crate::password::generated_secret_name(policy_name, &role.name, gen_spec);
1387                    names.insert(secret_name);
1388                }
1389            }
1390        }
1391        names
1392    }
1393}
1394
1395// ---------------------------------------------------------------------------
1396// Conversion: CRD spec → core manifest types
1397// ---------------------------------------------------------------------------
1398
1399impl PostgresPolicySpec {
1400    /// Convert the CRD spec into a `PolicyManifest` for use with the core library.
1401    pub fn to_policy_manifest(&self) -> pgroles_core::manifest::PolicyManifest {
1402        use pgroles_core::manifest::{
1403            DefaultPrivilegeGrant, MemberSpec, PolicyManifest, Profile, ProfileGrant,
1404            ProfileObjectTarget, RoleDefinition,
1405        };
1406
1407        let profiles = self
1408            .profiles
1409            .iter()
1410            .map(|(name, spec)| {
1411                let profile = Profile {
1412                    login: spec.login,
1413                    inherit: spec.inherit,
1414                    grants: spec
1415                        .grants
1416                        .iter()
1417                        .map(|g| ProfileGrant {
1418                            privileges: g.privileges.clone(),
1419                            object: ProfileObjectTarget {
1420                                object_type: g.object.object_type,
1421                                name: g.object.name.clone(),
1422                            },
1423                        })
1424                        .collect(),
1425                    default_privileges: spec
1426                        .default_privileges
1427                        .iter()
1428                        .map(|dp| DefaultPrivilegeGrant {
1429                            role: dp.role.clone(),
1430                            privileges: dp.privileges.clone(),
1431                            on_type: dp.on_type,
1432                        })
1433                        .collect(),
1434                };
1435                (name.clone(), profile)
1436            })
1437            .collect();
1438
1439        let roles = self
1440            .roles
1441            .iter()
1442            .map(|r| RoleDefinition {
1443                name: r.name.clone(),
1444                external: r.external,
1445                login: r.login,
1446                superuser: r.superuser,
1447                createdb: r.createdb,
1448                createrole: r.createrole,
1449                inherit: r.inherit,
1450                replication: r.replication,
1451                bypassrls: r.bypassrls,
1452                connection_limit: r.connection_limit,
1453                comment: r.comment.clone(),
1454                password: None, // K8s passwords are resolved separately via Secret refs
1455                password_valid_until: r.password_valid_until.clone(),
1456            })
1457            .collect();
1458
1459        // Memberships need MemberSpec conversion — the core type should
1460        // already be compatible since we use it directly in the CRD spec.
1461        // But we need to ensure the serde aliases work. Let's rebuild to be safe.
1462        let memberships = self
1463            .memberships
1464            .iter()
1465            .map(|m| pgroles_core::manifest::Membership {
1466                role: m.role.clone(),
1467                members: m
1468                    .members
1469                    .iter()
1470                    .map(|ms| MemberSpec {
1471                        name: ms.name.clone(),
1472                        inherit: ms.inherit,
1473                        admin: ms.admin,
1474                    })
1475                    .collect(),
1476            })
1477            .collect();
1478
1479        PolicyManifest {
1480            default_owner: self.default_owner.clone(),
1481            auth_providers: Vec::new(),
1482            profiles,
1483            schemas: self.schemas.clone(),
1484            roles,
1485            grants: self.grants.clone(),
1486            default_privileges: self.default_privileges.clone(),
1487            memberships,
1488            retirements: self.retirements.clone(),
1489        }
1490    }
1491
1492    /// Derive a conservative ownership claim set from the policy spec.
1493    ///
1494    /// This intentionally claims all declared/expanded roles and all referenced
1495    /// schemas so overlapping policies are rejected safely.
1496    pub fn ownership_claims(
1497        &self,
1498    ) -> Result<OwnershipClaims, pgroles_core::manifest::ManifestError> {
1499        let manifest = self.to_policy_manifest();
1500        let expanded = pgroles_core::manifest::expand_manifest(&manifest)?;
1501
1502        let mut roles: BTreeSet<String> = expanded.roles.into_iter().map(|r| r.name).collect();
1503        let mut schemas: BTreeSet<String> = self.schemas.iter().map(|s| s.name.clone()).collect();
1504
1505        roles.extend(manifest.retirements.into_iter().map(|r| r.role));
1506        roles.extend(manifest.grants.iter().map(|g| g.role.clone()));
1507        roles.extend(
1508            manifest
1509                .default_privileges
1510                .iter()
1511                .flat_map(|dp| dp.grant.iter().filter_map(|grant| grant.role.clone())),
1512        );
1513        roles.extend(manifest.memberships.iter().map(|m| m.role.clone()));
1514        roles.extend(
1515            manifest
1516                .memberships
1517                .iter()
1518                .flat_map(|m| m.members.iter().map(|member| member.name.clone())),
1519        );
1520
1521        schemas.extend(
1522            manifest
1523                .grants
1524                .iter()
1525                .filter_map(|g| match g.object.object_type {
1526                    ObjectType::Database => None,
1527                    ObjectType::Schema => g.object.name.clone(),
1528                    _ => g.object.schema.clone(),
1529                }),
1530        );
1531        schemas.extend(
1532            manifest
1533                .default_privileges
1534                .iter()
1535                .map(|dp| dp.schema.clone()),
1536        );
1537
1538        Ok(OwnershipClaims { roles, schemas })
1539    }
1540}
1541
1542// ---------------------------------------------------------------------------
1543// Status helpers
1544// ---------------------------------------------------------------------------
1545
1546impl PostgresPolicyStatus {
1547    /// Set a condition, replacing any existing condition of the same type.
1548    ///
1549    /// If the condition's `status` value has not changed, the existing
1550    /// `last_transition_time` is preserved (per Kubernetes condition conventions).
1551    pub fn set_condition(&mut self, new: PolicyCondition) {
1552        if let Some(existing) = self
1553            .conditions
1554            .iter()
1555            .find(|c| c.condition_type == new.condition_type)
1556            && existing.status == new.status
1557        {
1558            // Status unchanged — preserve the existing transition time.
1559            let mut updated = new;
1560            updated.last_transition_time = existing.last_transition_time.clone();
1561            self.conditions
1562                .retain(|c| c.condition_type != updated.condition_type);
1563            self.conditions.push(updated);
1564            return;
1565        }
1566        // New condition or status changed — use the new timestamp.
1567        self.conditions
1568            .retain(|c| c.condition_type != new.condition_type);
1569        self.conditions.push(new);
1570    }
1571}
1572
1573/// Create a timestamp string in ISO 8601 / RFC 3339 format.
1574pub fn now_rfc3339() -> String {
1575    // Use k8s-openapi's chrono re-export or manual formatting.
1576    // For simplicity, use the system time.
1577    use std::time::SystemTime;
1578    let now = SystemTime::now()
1579        .duration_since(SystemTime::UNIX_EPOCH)
1580        .unwrap_or_default();
1581    // Format as simplified ISO 8601
1582    let secs = now.as_secs();
1583    let days = secs / 86400;
1584    let remaining = secs % 86400;
1585    let hours = remaining / 3600;
1586    let minutes = (remaining % 3600) / 60;
1587    let seconds = remaining % 60;
1588
1589    // Convert days since epoch to date (simplified — good enough for status)
1590    let (year, month, day) = days_to_date(days);
1591    format!("{year:04}-{month:02}-{day:02}T{hours:02}:{minutes:02}:{seconds:02}Z")
1592}
1593
1594/// Convert days since Unix epoch to (year, month, day).
1595pub fn days_to_date(days_since_epoch: u64) -> (u64, u64, u64) {
1596    // Civil calendar algorithm from Howard Hinnant
1597    let z = days_since_epoch + 719468;
1598    let era = z / 146097;
1599    let doe = z - era * 146097;
1600    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
1601    let y = yoe + era * 400;
1602    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
1603    let mp = (5 * doy + 2) / 153;
1604    let d = doy - (153 * mp + 2) / 5 + 1;
1605    let m = if mp < 10 { mp + 3 } else { mp - 9 };
1606    let y = if m <= 2 { y + 1 } else { y };
1607    (y, m, d)
1608}
1609
1610/// Helper to create a "Ready" condition.
1611pub fn ready_condition(status: bool, reason: &str, message: &str) -> PolicyCondition {
1612    PolicyCondition {
1613        condition_type: "Ready".to_string(),
1614        status: if status { "True" } else { "False" }.to_string(),
1615        reason: Some(reason.to_string()),
1616        message: Some(message.to_string()),
1617        last_transition_time: Some(now_rfc3339()),
1618    }
1619}
1620
1621/// Helper to create a "Reconciling" condition.
1622pub fn reconciling_condition(message: &str) -> PolicyCondition {
1623    PolicyCondition {
1624        condition_type: "Reconciling".to_string(),
1625        status: "True".to_string(),
1626        reason: Some("Reconciling".to_string()),
1627        message: Some(message.to_string()),
1628        last_transition_time: Some(now_rfc3339()),
1629    }
1630}
1631
1632/// Helper to create a "Degraded" condition.
1633pub fn degraded_condition(reason: &str, message: &str) -> PolicyCondition {
1634    PolicyCondition {
1635        condition_type: "Degraded".to_string(),
1636        status: "True".to_string(),
1637        reason: Some(reason.to_string()),
1638        message: Some(message.to_string()),
1639        last_transition_time: Some(now_rfc3339()),
1640    }
1641}
1642
1643/// Helper to create a "Paused" condition.
1644pub fn paused_condition(message: &str) -> PolicyCondition {
1645    PolicyCondition {
1646        condition_type: "Paused".to_string(),
1647        status: "True".to_string(),
1648        reason: Some("Suspended".to_string()),
1649        message: Some(message.to_string()),
1650        last_transition_time: Some(now_rfc3339()),
1651    }
1652}
1653
1654/// Helper to create a "Conflict" condition.
1655pub fn conflict_condition(reason: &str, message: &str) -> PolicyCondition {
1656    PolicyCondition {
1657        condition_type: "Conflict".to_string(),
1658        status: "True".to_string(),
1659        reason: Some(reason.to_string()),
1660        message: Some(message.to_string()),
1661        last_transition_time: Some(now_rfc3339()),
1662    }
1663}
1664
1665/// Helper to create a "Drifted" condition.
1666pub fn drifted_condition(status: bool, reason: &str, message: &str) -> PolicyCondition {
1667    PolicyCondition {
1668        condition_type: "Drifted".to_string(),
1669        status: if status { "True" } else { "False" }.to_string(),
1670        reason: Some(reason.to_string()),
1671        message: Some(message.to_string()),
1672        last_transition_time: Some(now_rfc3339()),
1673    }
1674}
1675
1676// ---------------------------------------------------------------------------
1677// Tests
1678// ---------------------------------------------------------------------------
1679
1680#[cfg(test)]
1681mod tests {
1682    use super::*;
1683    use kube::CustomResourceExt;
1684
1685    #[test]
1686    fn crd_generates_valid_schema() {
1687        let crd = PostgresPolicy::crd();
1688        let yaml = serde_yaml::to_string(&crd).expect("CRD should serialize to YAML");
1689        assert!(yaml.contains("pgroles.io"), "group should be pgroles.io");
1690        assert!(yaml.contains("v1alpha1"), "version should be v1alpha1");
1691        assert!(
1692            yaml.contains("PostgresPolicy"),
1693            "kind should be PostgresPolicy"
1694        );
1695        assert!(
1696            yaml.contains("\"mode\"") || yaml.contains(" mode:"),
1697            "schema should declare spec.mode"
1698        );
1699        assert!(
1700            yaml.contains("\"object\"") || yaml.contains(" object:"),
1701            "schema should declare grant object targets using object"
1702        );
1703    }
1704
1705    #[test]
1706    fn spec_to_policy_manifest_roundtrip() {
1707        let spec = PostgresPolicySpec {
1708            connection: ConnectionSpec {
1709                secret_ref: Some(SecretReference {
1710                    name: "pg-secret".to_string(),
1711                }),
1712                secret_key: Some("DATABASE_URL".to_string()),
1713                params: None,
1714            },
1715            interval: "5m".to_string(),
1716            suspend: false,
1717            mode: PolicyMode::Apply,
1718            reconciliation_mode: CrdReconciliationMode::default(),
1719            default_owner: Some("app_owner".to_string()),
1720            profiles: std::collections::HashMap::new(),
1721            schemas: vec![],
1722            roles: vec![RoleSpec {
1723                name: "analytics".to_string(),
1724                external: true,
1725                login: Some(true),
1726                superuser: None,
1727                createdb: None,
1728                createrole: None,
1729                inherit: None,
1730                replication: None,
1731                bypassrls: None,
1732                connection_limit: None,
1733                comment: Some("test role".to_string()),
1734                password: None,
1735                password_valid_until: None,
1736            }],
1737            grants: vec![],
1738            default_privileges: vec![],
1739            memberships: vec![],
1740            retirements: vec![RoleRetirement {
1741                role: "legacy-app".to_string(),
1742                reassign_owned_to: Some("app_owner".to_string()),
1743                drop_owned: true,
1744                terminate_sessions: true,
1745            }],
1746            approval: None,
1747        };
1748
1749        let manifest = spec.to_policy_manifest();
1750        assert_eq!(manifest.default_owner, Some("app_owner".to_string()));
1751        assert_eq!(manifest.roles.len(), 1);
1752        assert_eq!(manifest.roles[0].name, "analytics");
1753        assert!(manifest.roles[0].external);
1754        assert_eq!(manifest.roles[0].login, Some(true));
1755        assert_eq!(manifest.roles[0].comment, Some("test role".to_string()));
1756        assert_eq!(manifest.retirements.len(), 1);
1757        assert_eq!(manifest.retirements[0].role, "legacy-app");
1758        assert_eq!(
1759            manifest.retirements[0].reassign_owned_to.as_deref(),
1760            Some("app_owner")
1761        );
1762        assert!(manifest.retirements[0].drop_owned);
1763        assert!(manifest.retirements[0].terminate_sessions);
1764    }
1765
1766    #[test]
1767    fn spec_to_policy_manifest_preserves_profile_inherit() {
1768        let spec = PostgresPolicySpec {
1769            connection: ConnectionSpec {
1770                secret_ref: Some(SecretReference {
1771                    name: "pg-secret".to_string(),
1772                }),
1773                secret_key: Some("DATABASE_URL".to_string()),
1774                params: None,
1775            },
1776            interval: "5m".to_string(),
1777            suspend: false,
1778            mode: PolicyMode::Apply,
1779            reconciliation_mode: CrdReconciliationMode::default(),
1780            default_owner: None,
1781            profiles: std::collections::HashMap::from([(
1782                "editor".to_string(),
1783                ProfileSpec {
1784                    login: Some(false),
1785                    inherit: Some(false),
1786                    grants: vec![],
1787                    default_privileges: vec![],
1788                },
1789            )]),
1790            schemas: vec![],
1791            roles: vec![],
1792            grants: vec![],
1793            default_privileges: vec![],
1794            memberships: vec![],
1795            retirements: vec![],
1796            approval: None,
1797        };
1798
1799        let manifest = spec.to_policy_manifest();
1800        assert_eq!(manifest.profiles["editor"].login, Some(false));
1801        assert_eq!(manifest.profiles["editor"].inherit, Some(false));
1802    }
1803
1804    #[test]
1805    fn status_set_condition_replaces_existing() {
1806        let mut status = PostgresPolicyStatus::default();
1807
1808        status.set_condition(ready_condition(false, "Pending", "Initial"));
1809        assert_eq!(status.conditions.len(), 1);
1810        assert_eq!(status.conditions[0].status, "False");
1811
1812        status.set_condition(ready_condition(true, "Reconciled", "All good"));
1813        assert_eq!(status.conditions.len(), 1);
1814        assert_eq!(status.conditions[0].status, "True");
1815        assert_eq!(status.conditions[0].reason.as_deref(), Some("Reconciled"));
1816    }
1817
1818    #[test]
1819    fn status_set_condition_adds_new_type() {
1820        let mut status = PostgresPolicyStatus::default();
1821
1822        status.set_condition(ready_condition(true, "OK", "ready"));
1823        status.set_condition(degraded_condition("Error", "something broke"));
1824
1825        assert_eq!(status.conditions.len(), 2);
1826    }
1827
1828    #[test]
1829    fn paused_condition_has_expected_shape() {
1830        let paused = paused_condition("paused by spec");
1831        assert_eq!(paused.condition_type, "Paused");
1832        assert_eq!(paused.status, "True");
1833        assert_eq!(paused.reason.as_deref(), Some("Suspended"));
1834    }
1835
1836    #[test]
1837    fn ownership_claims_include_expanded_roles_and_schemas() {
1838        let mut profiles = std::collections::HashMap::new();
1839        profiles.insert(
1840            "editor".to_string(),
1841            ProfileSpec {
1842                login: Some(false),
1843                inherit: None,
1844                grants: vec![],
1845                default_privileges: vec![],
1846            },
1847        );
1848
1849        let spec = PostgresPolicySpec {
1850            connection: ConnectionSpec {
1851                secret_ref: Some(SecretReference {
1852                    name: "pg-secret".to_string(),
1853                }),
1854                secret_key: Some("DATABASE_URL".to_string()),
1855                params: None,
1856            },
1857            interval: "5m".to_string(),
1858            suspend: false,
1859            mode: PolicyMode::Apply,
1860            reconciliation_mode: CrdReconciliationMode::default(),
1861            default_owner: None,
1862            profiles,
1863            schemas: vec![SchemaBinding {
1864                name: "inventory".to_string(),
1865                profiles: vec!["editor".to_string()],
1866                role_pattern: "{schema}-{profile}".to_string(),
1867                owner: None,
1868            }],
1869            roles: vec![RoleSpec {
1870                name: "app-service".to_string(),
1871                external: false,
1872                login: Some(true),
1873                superuser: None,
1874                createdb: None,
1875                createrole: None,
1876                inherit: None,
1877                replication: None,
1878                bypassrls: None,
1879                connection_limit: None,
1880                comment: None,
1881                password: None,
1882                password_valid_until: None,
1883            }],
1884            grants: vec![],
1885            default_privileges: vec![],
1886            memberships: vec![],
1887            retirements: vec![RoleRetirement {
1888                role: "legacy-app".to_string(),
1889                reassign_owned_to: None,
1890                drop_owned: false,
1891                terminate_sessions: false,
1892            }],
1893            approval: None,
1894        };
1895
1896        let claims = spec.ownership_claims().unwrap();
1897        assert!(claims.roles.contains("inventory-editor"));
1898        assert!(claims.roles.contains("app-service"));
1899        assert!(claims.roles.contains("legacy-app"));
1900        assert!(claims.schemas.contains("inventory"));
1901    }
1902
1903    #[test]
1904    fn ownership_overlap_summary_reports_roles_and_schemas() {
1905        let mut left = OwnershipClaims::default();
1906        left.roles.insert("analytics".to_string());
1907        left.schemas.insert("reporting".to_string());
1908
1909        let mut right = OwnershipClaims::default();
1910        right.roles.insert("analytics".to_string());
1911        right.schemas.insert("reporting".to_string());
1912        right.schemas.insert("other".to_string());
1913
1914        assert!(left.overlaps(&right));
1915        let summary = left.overlap_summary(&right);
1916        assert!(summary.contains("roles: analytics"));
1917        assert!(summary.contains("schemas: reporting"));
1918    }
1919
1920    #[test]
1921    fn database_identity_uses_namespace_and_identity_key() {
1922        let conn = ConnectionSpec {
1923            secret_ref: Some(SecretReference {
1924                name: "db-creds".to_string(),
1925            }),
1926            secret_key: Some("DATABASE_URL".to_string()),
1927            params: None,
1928        };
1929        let identity = DatabaseIdentity::from_connection("prod", &conn);
1930        assert_eq!(identity.as_str(), "prod/db-creds/DATABASE_URL");
1931    }
1932
1933    #[test]
1934    fn identity_key_same_database_different_users_are_equal() {
1935        // Two policies targeting the same database but with different users
1936        // should have the SAME identity key (for locking/conflict detection).
1937        let user_a = ConnectionSpec {
1938            secret_ref: None,
1939            secret_key: None,
1940            params: Some(ConnectionParams {
1941                host: Some("my-host".into()),
1942                host_secret: None,
1943                port: None,
1944                port_secret: None,
1945                dbname: Some("mydb".into()),
1946                dbname_secret: None,
1947                username: Some("alice".into()),
1948                username_secret: None,
1949                password: Some("pass-a".into()),
1950                password_secret: None,
1951                auth: None,
1952                ssl_mode: None,
1953                ssl_mode_secret: None,
1954                set_role: None,
1955            }),
1956        };
1957        let user_b = ConnectionSpec {
1958            secret_ref: None,
1959            secret_key: None,
1960            params: Some(ConnectionParams {
1961                host: Some("my-host".into()),
1962                host_secret: None,
1963                port: None,
1964                port_secret: None,
1965                dbname: Some("mydb".into()),
1966                dbname_secret: None,
1967                username: Some("bob".into()),
1968                username_secret: None,
1969                password: Some("pass-b".into()),
1970                password_secret: None,
1971                auth: None,
1972                ssl_mode: None,
1973                ssl_mode_secret: None,
1974                set_role: None,
1975            }),
1976        };
1977
1978        assert_eq!(
1979            user_a.identity_key(),
1980            user_b.identity_key(),
1981            "same database with different users should have the same identity key"
1982        );
1983        // But cache keys should differ (different credentials = different pool).
1984        assert_ne!(
1985            user_a.cache_key("default"),
1986            user_b.cache_key("default"),
1987            "different credentials should produce different cache keys"
1988        );
1989    }
1990
1991    #[test]
1992    fn cache_key_no_collision_between_literal_and_secret_username() {
1993        // A literal username containing "secret=" should not collide with a
1994        // real secret reference in the cache key.
1995        let literal_conn = ConnectionSpec {
1996            secret_ref: None,
1997            secret_key: None,
1998            params: Some(ConnectionParams {
1999                host: Some("my-host".into()),
2000                host_secret: None,
2001                port: None,
2002                port_secret: None,
2003                dbname: Some("mydb".into()),
2004                dbname_secret: None,
2005                username: Some("secret=creds\0password".into()),
2006                username_secret: None,
2007                password: Some("pass".into()),
2008                password_secret: None,
2009                auth: None,
2010                ssl_mode: None,
2011                ssl_mode_secret: None,
2012                set_role: None,
2013            }),
2014        };
2015        let secret_conn = ConnectionSpec {
2016            secret_ref: None,
2017            secret_key: None,
2018            params: Some(ConnectionParams {
2019                host: Some("my-host".into()),
2020                host_secret: None,
2021                port: None,
2022                port_secret: None,
2023                dbname: Some("mydb".into()),
2024                dbname_secret: None,
2025                username: None,
2026                username_secret: Some(SecretKeySelector {
2027                    name: "creds".into(),
2028                    key: "password".into(),
2029                }),
2030                password: Some("pass".into()),
2031                password_secret: None,
2032                auth: None,
2033                ssl_mode: None,
2034                ssl_mode_secret: None,
2035                set_role: None,
2036            }),
2037        };
2038
2039        assert_ne!(
2040            literal_conn.cache_key("default"),
2041            secret_conn.cache_key("default"),
2042            "literal and secret ref should produce different cache keys"
2043        );
2044    }
2045
2046    #[test]
2047    fn cache_key_includes_ssl_mode() {
2048        let conn_no_ssl = ConnectionSpec {
2049            secret_ref: None,
2050            secret_key: None,
2051            params: Some(ConnectionParams {
2052                host: Some("host".into()),
2053                host_secret: None,
2054                port: None,
2055                port_secret: None,
2056                dbname: Some("db".into()),
2057                dbname_secret: None,
2058                username: Some("user".into()),
2059                username_secret: None,
2060                password: Some("pass".into()),
2061                password_secret: None,
2062                auth: None,
2063                ssl_mode: None,
2064                ssl_mode_secret: None,
2065                set_role: None,
2066            }),
2067        };
2068        let conn_with_ssl = ConnectionSpec {
2069            secret_ref: None,
2070            secret_key: None,
2071            params: Some(ConnectionParams {
2072                host: Some("host".into()),
2073                host_secret: None,
2074                port: None,
2075                port_secret: None,
2076                dbname: Some("db".into()),
2077                dbname_secret: None,
2078                username: Some("user".into()),
2079                username_secret: None,
2080                password: Some("pass".into()),
2081                password_secret: None,
2082                auth: None,
2083                ssl_mode: Some("require".into()),
2084                ssl_mode_secret: None,
2085                set_role: None,
2086            }),
2087        };
2088
2089        assert_ne!(
2090            conn_no_ssl.cache_key("ns"),
2091            conn_with_ssl.cache_key("ns"),
2092            "cache key should differ when sslMode is present"
2093        );
2094    }
2095
2096    #[test]
2097    fn validate_connection_rejects_empty_literal_host() {
2098        let spec = spec_with_connection(ConnectionSpec {
2099            secret_ref: None,
2100            secret_key: None,
2101            params: Some(ConnectionParams {
2102                host: Some("".into()),
2103                host_secret: None,
2104                port: None,
2105                port_secret: None,
2106                dbname: Some("mydb".into()),
2107                dbname_secret: None,
2108                username: Some("user".into()),
2109                username_secret: None,
2110                password: Some("pass".into()),
2111                password_secret: None,
2112                auth: None,
2113                ssl_mode: None,
2114                ssl_mode_secret: None,
2115                set_role: None,
2116            }),
2117        });
2118
2119        let err = spec.validate_connection_spec().unwrap_err();
2120        assert!(
2121            matches!(err, ConnectionValidationError::EmptyLiteral { ref field } if field == "host"),
2122            "expected EmptyLiteral for host, got: {err}"
2123        );
2124    }
2125
2126    #[test]
2127    fn validate_connection_rejects_whitespace_literal_dbname() {
2128        let spec = spec_with_connection(ConnectionSpec {
2129            secret_ref: None,
2130            secret_key: None,
2131            params: Some(ConnectionParams {
2132                host: Some("host".into()),
2133                host_secret: None,
2134                port: None,
2135                port_secret: None,
2136                dbname: Some("  ".into()),
2137                dbname_secret: None,
2138                username: Some("user".into()),
2139                username_secret: None,
2140                password: Some("pass".into()),
2141                password_secret: None,
2142                auth: None,
2143                ssl_mode: None,
2144                ssl_mode_secret: None,
2145                set_role: None,
2146            }),
2147        });
2148
2149        let err = spec.validate_connection_spec().unwrap_err();
2150        assert!(
2151            matches!(err, ConnectionValidationError::EmptyLiteral { ref field } if field == "dbname"),
2152            "expected EmptyLiteral for dbname, got: {err}"
2153        );
2154    }
2155
2156    /// Helper to build a minimal spec with the given connection and no roles/grants.
2157    fn spec_with_connection(connection: ConnectionSpec) -> PostgresPolicySpec {
2158        PostgresPolicySpec {
2159            connection,
2160            interval: "5m".into(),
2161            suspend: false,
2162            mode: PolicyMode::Apply,
2163            reconciliation_mode: CrdReconciliationMode::default(),
2164            default_owner: None,
2165            profiles: Default::default(),
2166            schemas: vec![],
2167            roles: vec![],
2168            grants: vec![],
2169            default_privileges: vec![],
2170            memberships: vec![],
2171            retirements: vec![],
2172            approval: None,
2173        }
2174    }
2175
2176    fn url_mode_connection() -> ConnectionSpec {
2177        ConnectionSpec {
2178            secret_ref: Some(SecretReference {
2179                name: "pg-creds".into(),
2180            }),
2181            secret_key: Some("DATABASE_URL".into()),
2182            params: None,
2183        }
2184    }
2185
2186    fn params_mode_connection() -> ConnectionSpec {
2187        ConnectionSpec {
2188            secret_ref: None,
2189            secret_key: None,
2190            params: Some(ConnectionParams {
2191                host: Some("my-postgres".into()),
2192                host_secret: None,
2193                port: None,
2194                port_secret: None,
2195                dbname: Some("mydb".into()),
2196                dbname_secret: None,
2197                username: None,
2198                username_secret: Some(SecretKeySelector {
2199                    name: "pg-creds".into(),
2200                    key: "username".into(),
2201                }),
2202                password: None,
2203                password_secret: Some(SecretKeySelector {
2204                    name: "pg-creds".into(),
2205                    key: "password".into(),
2206                }),
2207                auth: None,
2208                ssl_mode: None,
2209                ssl_mode_secret: None,
2210                set_role: None,
2211            }),
2212        }
2213    }
2214
2215    // -- Connection validation tests -----------------------------------------
2216
2217    #[test]
2218    fn validate_connection_accepts_url_mode() {
2219        let spec = spec_with_connection(url_mode_connection());
2220        assert!(spec.validate_connection_spec().is_ok());
2221    }
2222
2223    #[test]
2224    fn validate_connection_accepts_params_mode() {
2225        let spec = spec_with_connection(params_mode_connection());
2226        assert!(spec.validate_connection_spec().is_ok());
2227    }
2228
2229    #[test]
2230    fn validate_connection_rejects_both_modes_set() {
2231        let spec = spec_with_connection(ConnectionSpec {
2232            secret_ref: Some(SecretReference {
2233                name: "pg-creds".into(),
2234            }),
2235            secret_key: None,
2236            params: Some(ConnectionParams {
2237                host: Some("host".into()),
2238                host_secret: None,
2239                port: None,
2240                port_secret: None,
2241                dbname: Some("db".into()),
2242                dbname_secret: None,
2243                username: Some("user".into()),
2244                username_secret: None,
2245                password: Some("pass".into()),
2246                password_secret: None,
2247                auth: None,
2248                ssl_mode: None,
2249                ssl_mode_secret: None,
2250                set_role: None,
2251            }),
2252        });
2253        assert!(matches!(
2254            spec.validate_connection_spec(),
2255            Err(ConnectionValidationError::BothModesSet)
2256        ));
2257    }
2258
2259    #[test]
2260    fn validate_connection_rejects_neither_mode_set() {
2261        let spec = spec_with_connection(ConnectionSpec {
2262            secret_ref: None,
2263            secret_key: None,
2264            params: None,
2265        });
2266        assert!(spec.validate_connection_spec().is_err());
2267    }
2268
2269    #[test]
2270    fn validate_connection_rejects_invalid_ssl_mode() {
2271        let spec = spec_with_connection(ConnectionSpec {
2272            secret_ref: None,
2273            secret_key: None,
2274            params: Some(ConnectionParams {
2275                host: Some("host".into()),
2276                host_secret: None,
2277                port: None,
2278                port_secret: None,
2279                dbname: Some("db".into()),
2280                dbname_secret: None,
2281                username: Some("user".into()),
2282                username_secret: None,
2283                password: Some("pass".into()),
2284                password_secret: None,
2285                auth: None,
2286                ssl_mode: Some("invalid-mode".into()),
2287                ssl_mode_secret: None,
2288                set_role: None,
2289            }),
2290        });
2291        assert!(spec.validate_connection_spec().is_err());
2292    }
2293
2294    fn params_with_set_role(set_role: Option<String>) -> ConnectionParams {
2295        ConnectionParams {
2296            host: Some("host".into()),
2297            host_secret: None,
2298            port: None,
2299            port_secret: None,
2300            dbname: Some("db".into()),
2301            dbname_secret: None,
2302            username: Some("user".into()),
2303            username_secret: None,
2304            password: Some("pass".into()),
2305            password_secret: None,
2306            auth: None,
2307            ssl_mode: None,
2308            ssl_mode_secret: None,
2309            set_role,
2310        }
2311    }
2312
2313    #[test]
2314    fn validate_connection_accepts_valid_set_role() {
2315        for role in [
2316            "cloudsqlsuperuser",
2317            "_underscore_start",
2318            "role-with-dash",
2319            "role_with$dollar",
2320            "Mixed_Case_Role",
2321            "r2d2",
2322        ] {
2323            let spec = spec_with_connection(ConnectionSpec {
2324                secret_ref: None,
2325                secret_key: None,
2326                params: Some(params_with_set_role(Some(role.into()))),
2327            });
2328            assert!(
2329                spec.validate_connection_spec().is_ok(),
2330                "expected {role} to be accepted"
2331            );
2332        }
2333    }
2334
2335    #[test]
2336    fn validate_connection_rejects_invalid_set_role() {
2337        for role in [
2338            "1leading_digit",
2339            "has space",
2340            "has\"quote",
2341            "has;semicolon",
2342            "has'singlequote",
2343            "ünicode",
2344        ] {
2345            let spec = spec_with_connection(ConnectionSpec {
2346                secret_ref: None,
2347                secret_key: None,
2348                params: Some(params_with_set_role(Some(role.into()))),
2349            });
2350            let err = spec
2351                .validate_connection_spec()
2352                .expect_err(&format!("expected {role} to be rejected"));
2353            assert!(
2354                matches!(err, ConnectionValidationError::InvalidRoleName { ref value } if value == role),
2355                "unexpected error for {role}: {err:?}",
2356            );
2357        }
2358    }
2359
2360    #[test]
2361    fn validate_connection_rejects_empty_set_role() {
2362        let spec = spec_with_connection(ConnectionSpec {
2363            secret_ref: None,
2364            secret_key: None,
2365            params: Some(params_with_set_role(Some("   ".into()))),
2366        });
2367        assert!(matches!(
2368            spec.validate_connection_spec(),
2369            Err(ConnectionValidationError::EmptyLiteral { ref field }) if field == "setRole"
2370        ));
2371    }
2372
2373    #[test]
2374    fn cache_key_includes_set_role() {
2375        let conn_no_role = ConnectionSpec {
2376            secret_ref: None,
2377            secret_key: None,
2378            params: Some(params_with_set_role(None)),
2379        };
2380        let conn_with_role = ConnectionSpec {
2381            secret_ref: None,
2382            secret_key: None,
2383            params: Some(params_with_set_role(Some("cloudsqlsuperuser".into()))),
2384        };
2385        assert_ne!(
2386            conn_no_role.cache_key("ns"),
2387            conn_with_role.cache_key("ns"),
2388            "cache key should differ when setRole is present"
2389        );
2390    }
2391
2392    #[test]
2393    fn validate_connection_accepts_gcp_workload_identity_without_password() {
2394        let spec = spec_with_connection(ConnectionSpec {
2395            secret_ref: None,
2396            secret_key: None,
2397            params: Some(ConnectionParams {
2398                host: Some("10.0.0.5".into()),
2399                host_secret: None,
2400                port: None,
2401                port_secret: None,
2402                dbname: Some("discovery".into()),
2403                dbname_secret: None,
2404                username: Some("pgroles-operator@my-project.iam".into()),
2405                username_secret: None,
2406                password: None,
2407                password_secret: None,
2408                auth: Some(ConnectionAuth::GcpWorkloadIdentity {
2409                    impersonate_service_account: None,
2410                    scope: None,
2411                }),
2412                ssl_mode: None,
2413                ssl_mode_secret: None,
2414                set_role: None,
2415            }),
2416        });
2417
2418        assert!(spec.validate_connection_spec().is_ok());
2419        assert!(spec.referenced_secret_names("policy").is_empty());
2420    }
2421
2422    #[test]
2423    fn validate_connection_rejects_gcp_workload_identity_with_password() {
2424        let spec = spec_with_connection(ConnectionSpec {
2425            secret_ref: None,
2426            secret_key: None,
2427            params: Some(ConnectionParams {
2428                host: Some("10.0.0.5".into()),
2429                host_secret: None,
2430                port: None,
2431                port_secret: None,
2432                dbname: Some("discovery".into()),
2433                dbname_secret: None,
2434                username: Some("pgroles-operator@my-project.iam".into()),
2435                username_secret: None,
2436                password: Some("static-password".into()),
2437                password_secret: None,
2438                auth: Some(ConnectionAuth::GcpWorkloadIdentity {
2439                    impersonate_service_account: None,
2440                    scope: None,
2441                }),
2442                ssl_mode: None,
2443                ssl_mode_secret: None,
2444                set_role: None,
2445            }),
2446        });
2447
2448        assert!(matches!(
2449            spec.validate_connection_spec(),
2450            Err(ConnectionValidationError::AuthWithPassword)
2451        ));
2452    }
2453
2454    #[test]
2455    fn validate_connection_rejects_empty_gcp_auth_fields() {
2456        let spec = spec_with_connection(ConnectionSpec {
2457            secret_ref: None,
2458            secret_key: None,
2459            params: Some(ConnectionParams {
2460                host: Some("10.0.0.5".into()),
2461                host_secret: None,
2462                port: None,
2463                port_secret: None,
2464                dbname: Some("discovery".into()),
2465                dbname_secret: None,
2466                username: Some("pgroles-operator@my-project.iam".into()),
2467                username_secret: None,
2468                password: None,
2469                password_secret: None,
2470                auth: Some(ConnectionAuth::GcpWorkloadIdentity {
2471                    impersonate_service_account: Some(" ".into()),
2472                    scope: None,
2473                }),
2474                ssl_mode: None,
2475                ssl_mode_secret: None,
2476                set_role: None,
2477            }),
2478        });
2479
2480        assert!(matches!(
2481            spec.validate_connection_spec(),
2482            Err(ConnectionValidationError::EmptyAuthField { ref field })
2483                if field == "impersonateServiceAccount"
2484        ));
2485    }
2486
2487    #[test]
2488    fn validate_connection_accepts_valid_ssl_modes() {
2489        for mode in &[
2490            "disable",
2491            "allow",
2492            "prefer",
2493            "require",
2494            "verify-ca",
2495            "verify-full",
2496        ] {
2497            let spec = spec_with_connection(ConnectionSpec {
2498                secret_ref: None,
2499                secret_key: None,
2500                params: Some(ConnectionParams {
2501                    host: Some("host".into()),
2502                    host_secret: None,
2503                    port: None,
2504                    port_secret: None,
2505                    dbname: Some("db".into()),
2506                    dbname_secret: None,
2507                    username: Some("user".into()),
2508                    username_secret: None,
2509                    password: Some("pass".into()),
2510                    password_secret: None,
2511                    auth: None,
2512                    ssl_mode: Some((*mode).into()),
2513                    ssl_mode_secret: None,
2514                    set_role: None,
2515                }),
2516            });
2517            assert!(
2518                spec.validate_connection_spec().is_ok(),
2519                "sslMode '{mode}' should be accepted"
2520            );
2521        }
2522    }
2523
2524    #[test]
2525    fn validate_connection_rejects_empty_secret_name() {
2526        let spec = spec_with_connection(ConnectionSpec {
2527            secret_ref: None,
2528            secret_key: None,
2529            params: Some(ConnectionParams {
2530                host: Some("host".into()),
2531                host_secret: None,
2532                port: None,
2533                port_secret: None,
2534                dbname: Some("db".into()),
2535                dbname_secret: None,
2536                username: None,
2537                username_secret: Some(SecretKeySelector {
2538                    name: "".into(),
2539                    key: "username".into(),
2540                }),
2541                password: Some("pass".into()),
2542                password_secret: None,
2543                auth: None,
2544                ssl_mode: None,
2545                ssl_mode_secret: None,
2546                set_role: None,
2547            }),
2548        });
2549        assert!(spec.validate_connection_spec().is_err());
2550    }
2551
2552    #[test]
2553    fn validate_connection_rejects_both_literal_and_secret_for_same_field() {
2554        let spec = spec_with_connection(ConnectionSpec {
2555            secret_ref: None,
2556            secret_key: None,
2557            params: Some(ConnectionParams {
2558                host: Some("host".into()),
2559                host_secret: Some(SecretKeySelector {
2560                    name: "s".into(),
2561                    key: "k".into(),
2562                }),
2563                port: None,
2564                port_secret: None,
2565                dbname: Some("db".into()),
2566                dbname_secret: None,
2567                username: Some("user".into()),
2568                username_secret: None,
2569                password: Some("pass".into()),
2570                password_secret: None,
2571                auth: None,
2572                ssl_mode: None,
2573                ssl_mode_secret: None,
2574                set_role: None,
2575            }),
2576        });
2577        assert!(matches!(
2578            spec.validate_connection_spec(),
2579            Err(ConnectionValidationError::BothFieldsSet { ref field }) if field == "host"
2580        ));
2581    }
2582
2583    #[test]
2584    fn validate_connection_rejects_neither_literal_nor_secret_for_required_field() {
2585        let spec = spec_with_connection(ConnectionSpec {
2586            secret_ref: None,
2587            secret_key: None,
2588            params: Some(ConnectionParams {
2589                host: None,
2590                host_secret: None,
2591                port: None,
2592                port_secret: None,
2593                dbname: Some("db".into()),
2594                dbname_secret: None,
2595                username: Some("user".into()),
2596                username_secret: None,
2597                password: Some("pass".into()),
2598                password_secret: None,
2599                auth: None,
2600                ssl_mode: None,
2601                ssl_mode_secret: None,
2602                set_role: None,
2603            }),
2604        });
2605        assert!(matches!(
2606            spec.validate_connection_spec(),
2607            Err(ConnectionValidationError::NeitherFieldSet { ref field }) if field == "host"
2608        ));
2609    }
2610
2611    // -- ConnectionSpec backward compatibility --------------------------------
2612
2613    #[test]
2614    fn connection_spec_backward_compat_url_mode() {
2615        // The old format with required secretRef should still deserialize.
2616        let yaml = r#"
2617secretRef:
2618  name: pg-creds
2619secretKey: DATABASE_URL
2620"#;
2621        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2622        assert!(conn.secret_ref.is_some());
2623        assert_eq!(conn.effective_secret_key(), "DATABASE_URL");
2624        assert!(conn.params.is_none());
2625    }
2626
2627    #[test]
2628    fn connection_spec_backward_compat_default_secret_key() {
2629        let yaml = r#"
2630secretRef:
2631  name: pg-creds
2632"#;
2633        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2634        assert_eq!(conn.effective_secret_key(), "DATABASE_URL");
2635    }
2636
2637    #[test]
2638    fn connection_spec_params_mode_deserializes_keycloak_style() {
2639        let yaml = r#"
2640params:
2641  host: my-postgres
2642  port: 5432
2643  dbname: mydb
2644  usernameSecret:
2645    name: creds
2646    key: username
2647  passwordSecret:
2648    name: creds
2649    key: password
2650  sslMode: require
2651"#;
2652        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2653        assert!(conn.secret_ref.is_none());
2654        let params = conn.params.unwrap();
2655        assert_eq!(params.host.as_deref(), Some("my-postgres"));
2656        assert_eq!(params.port, Some(5432));
2657        assert!(params.username_secret.is_some());
2658        assert_eq!(params.username_secret.as_ref().unwrap().name, "creds");
2659        assert_eq!(params.ssl_mode.as_deref(), Some("require"));
2660    }
2661
2662    #[test]
2663    fn connection_spec_params_mode_deserializes_gcp_workload_identity_auth() {
2664        let yaml = r#"
2665params:
2666  host: 10.0.0.5
2667  port: 5432
2668  dbname: discovery
2669  username: pgroles-operator@my-project.iam
2670  auth:
2671    type: gcp_workload_identity
2672    impersonateServiceAccount: target@other-project.iam.gserviceaccount.com
2673    scope: https://example.com/custom-scope
2674"#;
2675        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2676        let params = conn.params.as_ref().unwrap();
2677        let auth = params.auth.as_ref().expect("auth should deserialize");
2678
2679        assert!(params.password.is_none());
2680        assert_eq!(auth.gcp_scope(), "https://example.com/custom-scope");
2681        assert_eq!(
2682            auth.gcp_impersonate_service_account(),
2683            Some("target@other-project.iam.gserviceaccount.com")
2684        );
2685        assert!(conn.cache_key("prod").contains("gcp_workload_identity"));
2686
2687        let spec = spec_with_connection(conn);
2688        assert!(spec.validate_connection_spec().is_ok());
2689    }
2690
2691    #[test]
2692    fn connection_spec_params_mode_all_secrets() {
2693        // CNPG/PGO pattern — everything from one secret.
2694        let yaml = r#"
2695params:
2696  hostSecret:
2697    name: cluster-app
2698    key: host
2699  portSecret:
2700    name: cluster-app
2701    key: port
2702  dbnameSecret:
2703    name: cluster-app
2704    key: dbname
2705  usernameSecret:
2706    name: cluster-app
2707    key: user
2708  passwordSecret:
2709    name: cluster-app
2710    key: password
2711"#;
2712        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2713        let params = conn.params.unwrap();
2714        assert!(params.host.is_none());
2715        assert!(params.host_secret.is_some());
2716        assert_eq!(params.host_secret.as_ref().unwrap().name, "cluster-app");
2717        assert!(params.port.is_none());
2718        assert!(params.port_secret.is_some());
2719    }
2720
2721    // -- referenced_secret_names with params mode ----------------------------
2722
2723    #[test]
2724    fn referenced_secret_names_includes_params_secrets() {
2725        let spec = spec_with_connection(params_mode_connection());
2726        let names = spec.referenced_secret_names("test-policy");
2727        assert!(
2728            names.contains("pg-creds"),
2729            "should include the credential secret from params"
2730        );
2731    }
2732
2733    #[test]
2734    fn referenced_secret_names_deduplicates_across_modes() {
2735        // Same secret name used in both connection and password secretRef.
2736        let mut spec = spec_with_connection(params_mode_connection());
2737        spec.roles = vec![RoleSpec {
2738            name: "app".into(),
2739            external: false,
2740            login: Some(true),
2741            password: Some(PasswordSpec {
2742                secret_ref: Some(SecretReference {
2743                    name: "pg-creds".into(),
2744                }),
2745                secret_key: Some("app-password".into()),
2746                generate: None,
2747            }),
2748            password_valid_until: None,
2749            superuser: None,
2750            createdb: None,
2751            createrole: None,
2752            inherit: None,
2753            replication: None,
2754            bypassrls: None,
2755            connection_limit: None,
2756            comment: None,
2757        }];
2758        let names = spec.referenced_secret_names("test-policy");
2759        // pg-creds appears in both connection params and password — should be deduped.
2760        assert_eq!(
2761            names.iter().filter(|n| *n == "pg-creds").count(),
2762            1,
2763            "BTreeSet should deduplicate"
2764        );
2765    }
2766
2767    // -- ConnectionParams port default ---------------------------------------
2768
2769    #[test]
2770    fn connection_params_port_defaults_to_none() {
2771        let yaml = r#"
2772params:
2773  host: my-host
2774  dbname: mydb
2775  username: user
2776  password: pass
2777"#;
2778        let conn: ConnectionSpec = serde_yaml::from_str(yaml).unwrap();
2779        let params = conn.params.unwrap();
2780        assert!(
2781            params.port.is_none(),
2782            "port should default to None (resolved as 5432 at runtime)"
2783        );
2784        assert!(
2785            params.port_secret.is_none(),
2786            "portSecret should also default to None"
2787        );
2788    }
2789
2790    #[test]
2791    fn now_rfc3339_produces_valid_format() {
2792        let ts = now_rfc3339();
2793        // Should match YYYY-MM-DDTHH:MM:SSZ
2794        assert!(ts.len() == 20, "expected 20 chars, got {}: {ts}", ts.len());
2795        assert!(ts.ends_with('Z'), "should end with Z: {ts}");
2796        assert_eq!(&ts[4..5], "-", "should have dash at pos 4: {ts}");
2797        assert_eq!(&ts[10..11], "T", "should have T at pos 10: {ts}");
2798    }
2799
2800    #[test]
2801    fn ready_condition_true_has_expected_shape() {
2802        let cond = ready_condition(true, "Reconciled", "All changes applied");
2803        assert_eq!(cond.condition_type, "Ready");
2804        assert_eq!(cond.status, "True");
2805        assert_eq!(cond.reason.as_deref(), Some("Reconciled"));
2806        assert_eq!(cond.message.as_deref(), Some("All changes applied"));
2807        assert!(cond.last_transition_time.is_some());
2808    }
2809
2810    #[test]
2811    fn ready_condition_false_has_expected_shape() {
2812        let cond = ready_condition(false, "InvalidSpec", "bad manifest");
2813        assert_eq!(cond.condition_type, "Ready");
2814        assert_eq!(cond.status, "False");
2815        assert_eq!(cond.reason.as_deref(), Some("InvalidSpec"));
2816        assert_eq!(cond.message.as_deref(), Some("bad manifest"));
2817    }
2818
2819    #[test]
2820    fn degraded_condition_has_expected_shape() {
2821        let cond = degraded_condition("InvalidSpec", "expansion failed");
2822        assert_eq!(cond.condition_type, "Degraded");
2823        assert_eq!(cond.status, "True");
2824        assert_eq!(cond.reason.as_deref(), Some("InvalidSpec"));
2825        assert_eq!(cond.message.as_deref(), Some("expansion failed"));
2826        assert!(cond.last_transition_time.is_some());
2827    }
2828
2829    #[test]
2830    fn reconciling_condition_has_expected_shape() {
2831        let cond = reconciling_condition("Reconciliation in progress");
2832        assert_eq!(cond.condition_type, "Reconciling");
2833        assert_eq!(cond.status, "True");
2834        assert_eq!(cond.reason.as_deref(), Some("Reconciling"));
2835        assert_eq!(cond.message.as_deref(), Some("Reconciliation in progress"));
2836        assert!(cond.last_transition_time.is_some());
2837    }
2838
2839    #[test]
2840    fn conflict_condition_has_expected_shape() {
2841        let cond = conflict_condition("ConflictingPolicy", "overlaps with ns/other");
2842        assert_eq!(cond.condition_type, "Conflict");
2843        assert_eq!(cond.status, "True");
2844        assert_eq!(cond.reason.as_deref(), Some("ConflictingPolicy"));
2845        assert_eq!(cond.message.as_deref(), Some("overlaps with ns/other"));
2846        assert!(cond.last_transition_time.is_some());
2847    }
2848
2849    #[test]
2850    fn ownership_claims_no_overlap() {
2851        let mut left = OwnershipClaims::default();
2852        left.roles.insert("analytics".to_string());
2853        left.schemas.insert("reporting".to_string());
2854
2855        let mut right = OwnershipClaims::default();
2856        right.roles.insert("billing".to_string());
2857        right.schemas.insert("payments".to_string());
2858
2859        assert!(!left.overlaps(&right));
2860        let summary = left.overlap_summary(&right);
2861        assert!(summary.is_empty());
2862    }
2863
2864    #[test]
2865    fn ownership_claims_partial_role_overlap() {
2866        let mut left = OwnershipClaims::default();
2867        left.roles.insert("analytics".to_string());
2868        left.roles.insert("reporting-viewer".to_string());
2869
2870        let mut right = OwnershipClaims::default();
2871        right.roles.insert("analytics".to_string());
2872        right.roles.insert("other-role".to_string());
2873
2874        assert!(left.overlaps(&right));
2875        let summary = left.overlap_summary(&right);
2876        assert!(summary.contains("roles: analytics"));
2877        assert!(!summary.contains("schemas"));
2878    }
2879
2880    #[test]
2881    fn ownership_claims_empty_is_disjoint() {
2882        let left = OwnershipClaims::default();
2883        let right = OwnershipClaims::default();
2884        assert!(!left.overlaps(&right));
2885    }
2886
2887    #[test]
2888    fn database_identity_equality() {
2889        let conn_a = ConnectionSpec {
2890            secret_ref: Some(SecretReference {
2891                name: "db-creds".to_string(),
2892            }),
2893            secret_key: Some("DATABASE_URL".to_string()),
2894            params: None,
2895        };
2896        let a = DatabaseIdentity::from_connection("prod", &conn_a);
2897        let b = DatabaseIdentity::from_connection("prod", &conn_a);
2898        let c = DatabaseIdentity::from_connection("staging", &conn_a);
2899        assert_eq!(a, b);
2900        assert_ne!(a, c);
2901    }
2902
2903    #[test]
2904    fn database_identity_different_key() {
2905        let conn_a = ConnectionSpec {
2906            secret_ref: Some(SecretReference {
2907                name: "db-creds".to_string(),
2908            }),
2909            secret_key: Some("DATABASE_URL".to_string()),
2910            params: None,
2911        };
2912        let conn_b = ConnectionSpec {
2913            secret_ref: Some(SecretReference {
2914                name: "db-creds".to_string(),
2915            }),
2916            secret_key: Some("CUSTOM_URL".to_string()),
2917            params: None,
2918        };
2919        let a = DatabaseIdentity::from_connection("prod", &conn_a);
2920        let b = DatabaseIdentity::from_connection("prod", &conn_b);
2921        assert_ne!(a, b);
2922    }
2923
2924    #[test]
2925    fn status_default_has_empty_conditions() {
2926        let status = PostgresPolicyStatus::default();
2927        assert!(status.conditions.is_empty());
2928        assert!(status.observed_generation.is_none());
2929        assert!(status.last_attempted_generation.is_none());
2930        assert!(status.last_successful_reconcile_time.is_none());
2931        assert!(status.change_summary.is_none());
2932        assert!(status.managed_database_identity.is_none());
2933        assert!(status.owned_roles.is_empty());
2934        assert!(status.owned_schemas.is_empty());
2935        assert!(status.last_error.is_none());
2936        assert!(status.applied_password_source_versions.is_empty());
2937    }
2938
2939    #[test]
2940    fn status_degraded_workflow_sets_ready_false_and_degraded_true() {
2941        let mut status = PostgresPolicyStatus::default();
2942
2943        // Simulate a failed reconciliation: Ready=False + Degraded=True
2944        status.set_condition(ready_condition(false, "InvalidSpec", "bad manifest"));
2945        status.set_condition(degraded_condition("InvalidSpec", "bad manifest"));
2946        status
2947            .conditions
2948            .retain(|c| c.condition_type != "Reconciling" && c.condition_type != "Paused");
2949        status.change_summary = None;
2950        status.last_error = Some("bad manifest".to_string());
2951
2952        // Verify Ready=False
2953        let ready = status
2954            .conditions
2955            .iter()
2956            .find(|c| c.condition_type == "Ready")
2957            .expect("should have Ready condition");
2958        assert_eq!(ready.status, "False");
2959        assert_eq!(ready.reason.as_deref(), Some("InvalidSpec"));
2960
2961        // Verify Degraded=True
2962        let degraded = status
2963            .conditions
2964            .iter()
2965            .find(|c| c.condition_type == "Degraded")
2966            .expect("should have Degraded condition");
2967        assert_eq!(degraded.status, "True");
2968        assert_eq!(degraded.reason.as_deref(), Some("InvalidSpec"));
2969
2970        // Verify last_error is set
2971        assert_eq!(status.last_error.as_deref(), Some("bad manifest"));
2972    }
2973
2974    #[test]
2975    fn status_conflict_workflow() {
2976        let mut status = PostgresPolicyStatus::default();
2977
2978        // Simulate a conflict
2979        let msg = "policy ownership overlaps with staging/other on database target prod/db/URL";
2980        status.set_condition(ready_condition(false, "ConflictingPolicy", msg));
2981        status.set_condition(conflict_condition("ConflictingPolicy", msg));
2982        status.set_condition(degraded_condition("ConflictingPolicy", msg));
2983        status
2984            .conditions
2985            .retain(|c| c.condition_type != "Reconciling");
2986        status.last_error = Some(msg.to_string());
2987
2988        // Verify Conflict=True
2989        let conflict = status
2990            .conditions
2991            .iter()
2992            .find(|c| c.condition_type == "Conflict")
2993            .expect("should have Conflict condition");
2994        assert_eq!(conflict.status, "True");
2995        assert_eq!(conflict.reason.as_deref(), Some("ConflictingPolicy"));
2996
2997        // Verify Ready=False
2998        let ready = status
2999            .conditions
3000            .iter()
3001            .find(|c| c.condition_type == "Ready")
3002            .expect("should have Ready condition");
3003        assert_eq!(ready.status, "False");
3004
3005        // Verify Degraded=True
3006        let degraded = status
3007            .conditions
3008            .iter()
3009            .find(|c| c.condition_type == "Degraded")
3010            .expect("should have Degraded condition");
3011        assert_eq!(degraded.status, "True");
3012    }
3013
3014    #[test]
3015    fn status_successful_reconcile_records_generation_and_time() {
3016        let mut status = PostgresPolicyStatus::default();
3017        let generation = Some(3_i64);
3018        let summary = ChangeSummary {
3019            roles_created: 2,
3020            total: 2,
3021            ..Default::default()
3022        };
3023
3024        // Simulate a successful reconciliation
3025        status.set_condition(ready_condition(true, "Reconciled", "All changes applied"));
3026        status.conditions.retain(|c| {
3027            c.condition_type != "Reconciling"
3028                && c.condition_type != "Degraded"
3029                && c.condition_type != "Conflict"
3030                && c.condition_type != "Paused"
3031        });
3032        status.observed_generation = generation;
3033        status.last_attempted_generation = generation;
3034        status.last_successful_reconcile_time = Some(now_rfc3339());
3035        status.last_reconcile_time = Some(now_rfc3339());
3036        status.change_summary = Some(summary);
3037        status.last_error = None;
3038
3039        // Verify Ready=True
3040        let ready = status
3041            .conditions
3042            .iter()
3043            .find(|c| c.condition_type == "Ready")
3044            .expect("should have Ready condition");
3045        assert_eq!(ready.status, "True");
3046        assert_eq!(ready.reason.as_deref(), Some("Reconciled"));
3047
3048        // Verify generation recorded
3049        assert_eq!(status.observed_generation, Some(3));
3050        assert_eq!(status.last_attempted_generation, Some(3));
3051
3052        // Verify timestamps set
3053        assert!(status.last_successful_reconcile_time.is_some());
3054        assert!(status.last_reconcile_time.is_some());
3055
3056        // Verify summary
3057        let summary = status.change_summary.as_ref().unwrap();
3058        assert_eq!(summary.roles_created, 2);
3059        assert_eq!(summary.total, 2);
3060
3061        // Verify no error
3062        assert!(status.last_error.is_none());
3063
3064        // Verify no Degraded/Conflict/Paused/Reconciling conditions
3065        assert!(
3066            status
3067                .conditions
3068                .iter()
3069                .all(|c| c.condition_type != "Degraded"
3070                    && c.condition_type != "Conflict"
3071                    && c.condition_type != "Paused"
3072                    && c.condition_type != "Reconciling")
3073        );
3074    }
3075
3076    #[test]
3077    fn status_suspended_workflow() {
3078        let mut status = PostgresPolicyStatus::default();
3079        let generation = Some(2_i64);
3080
3081        // Simulate a suspended reconciliation
3082        status.set_condition(paused_condition("Reconciliation suspended by spec"));
3083        status.set_condition(ready_condition(
3084            false,
3085            "Suspended",
3086            "Reconciliation suspended by spec",
3087        ));
3088        status
3089            .conditions
3090            .retain(|c| c.condition_type != "Reconciling");
3091        status.last_attempted_generation = generation;
3092        status.last_error = None;
3093
3094        // Verify Paused=True
3095        let paused = status
3096            .conditions
3097            .iter()
3098            .find(|c| c.condition_type == "Paused")
3099            .expect("should have Paused condition");
3100        assert_eq!(paused.status, "True");
3101
3102        // Verify Ready=False with Suspended reason
3103        let ready = status
3104            .conditions
3105            .iter()
3106            .find(|c| c.condition_type == "Ready")
3107            .expect("should have Ready condition");
3108        assert_eq!(ready.status, "False");
3109        assert_eq!(ready.reason.as_deref(), Some("Suspended"));
3110
3111        // Verify no Reconciling condition
3112        assert!(
3113            !status
3114                .conditions
3115                .iter()
3116                .any(|c| c.condition_type == "Reconciling")
3117        );
3118    }
3119
3120    #[test]
3121    fn status_transitions_from_degraded_to_ready() {
3122        let mut status = PostgresPolicyStatus::default();
3123
3124        // First, set degraded state
3125        status.set_condition(ready_condition(false, "InvalidSpec", "error"));
3126        status.set_condition(degraded_condition("InvalidSpec", "error"));
3127        status.last_error = Some("error".to_string());
3128
3129        assert_eq!(status.conditions.len(), 2);
3130
3131        // Then, resolve to ready
3132        status.set_condition(ready_condition(true, "Reconciled", "All changes applied"));
3133        status.conditions.retain(|c| {
3134            c.condition_type != "Reconciling"
3135                && c.condition_type != "Degraded"
3136                && c.condition_type != "Conflict"
3137                && c.condition_type != "Paused"
3138        });
3139        status.last_error = None;
3140
3141        // Verify Ready=True
3142        let ready = status
3143            .conditions
3144            .iter()
3145            .find(|c| c.condition_type == "Ready")
3146            .expect("should have Ready condition");
3147        assert_eq!(ready.status, "True");
3148
3149        // Verify Degraded removed
3150        assert!(
3151            !status
3152                .conditions
3153                .iter()
3154                .any(|c| c.condition_type == "Degraded")
3155        );
3156
3157        // Verify only Ready condition remains
3158        assert_eq!(status.conditions.len(), 1);
3159
3160        // Verify error cleared
3161        assert!(status.last_error.is_none());
3162    }
3163
3164    #[test]
3165    fn change_summary_default_is_all_zero() {
3166        let summary = ChangeSummary::default();
3167        assert_eq!(summary.roles_created, 0);
3168        assert_eq!(summary.roles_altered, 0);
3169        assert_eq!(summary.roles_dropped, 0);
3170        assert_eq!(summary.sessions_terminated, 0);
3171        assert_eq!(summary.grants_added, 0);
3172        assert_eq!(summary.grants_revoked, 0);
3173        assert_eq!(summary.default_privileges_set, 0);
3174        assert_eq!(summary.default_privileges_revoked, 0);
3175        assert_eq!(summary.members_added, 0);
3176        assert_eq!(summary.members_removed, 0);
3177        assert_eq!(summary.total, 0);
3178    }
3179
3180    #[test]
3181    fn status_serializes_to_json() {
3182        let mut status = PostgresPolicyStatus::default();
3183        status.set_condition(ready_condition(true, "Reconciled", "done"));
3184        status.observed_generation = Some(5);
3185        status.managed_database_identity = Some("ns/secret/key".to_string());
3186        status.owned_roles = vec!["role-a".to_string(), "role-b".to_string()];
3187        status.owned_schemas = vec!["public".to_string()];
3188        status.change_summary = Some(ChangeSummary {
3189            roles_created: 1,
3190            total: 1,
3191            ..Default::default()
3192        });
3193
3194        let json = serde_json::to_string(&status).expect("should serialize");
3195        assert!(json.contains("\"Reconciled\""));
3196        assert!(json.contains("\"observed_generation\":5"));
3197        assert!(json.contains("\"role-a\""));
3198        assert!(json.contains("\"ns/secret/key\""));
3199    }
3200
3201    #[test]
3202    fn crd_spec_deserializes_from_yaml() {
3203        let yaml = r#"
3204connection:
3205  secretRef:
3206    name: pg-credentials
3207interval: "10m"
3208default_owner: app_owner
3209profiles:
3210  editor:
3211    grants:
3212      - privileges: [USAGE]
3213        object: { type: schema }
3214      - privileges: [SELECT, INSERT, UPDATE, DELETE]
3215        object: { type: table, name: "*" }
3216    default_privileges:
3217      - privileges: [SELECT, INSERT, UPDATE, DELETE]
3218        on_type: table
3219schemas:
3220  - name: inventory
3221    profiles: [editor]
3222roles:
3223  - name: analytics
3224    login: true
3225grants:
3226  - role: analytics
3227    privileges: [CONNECT]
3228    object: { type: database, name: mydb }
3229memberships:
3230  - role: inventory-editor
3231    members:
3232      - name: analytics
3233retirements:
3234  - role: legacy-app
3235    reassign_owned_to: app_owner
3236    drop_owned: true
3237    terminate_sessions: true
3238"#;
3239        let spec: PostgresPolicySpec = serde_yaml::from_str(yaml).expect("should deserialize");
3240        assert_eq!(spec.interval, "10m");
3241        assert_eq!(spec.default_owner, Some("app_owner".to_string()));
3242        assert_eq!(spec.profiles.len(), 1);
3243        assert!(spec.profiles.contains_key("editor"));
3244        assert_eq!(spec.schemas.len(), 1);
3245        assert_eq!(spec.roles.len(), 1);
3246        assert_eq!(spec.grants.len(), 1);
3247        assert_eq!(spec.memberships.len(), 1);
3248        assert_eq!(spec.retirements.len(), 1);
3249        assert_eq!(spec.retirements[0].role, "legacy-app");
3250        assert!(spec.retirements[0].terminate_sessions);
3251    }
3252
3253    #[test]
3254    fn referenced_secret_names_includes_connection_secret() {
3255        let spec = PostgresPolicySpec {
3256            connection: ConnectionSpec {
3257                secret_ref: Some(SecretReference {
3258                    name: "pg-conn".to_string(),
3259                }),
3260                secret_key: Some("DATABASE_URL".to_string()),
3261                params: None,
3262            },
3263            interval: "5m".to_string(),
3264            suspend: false,
3265            mode: PolicyMode::Apply,
3266            reconciliation_mode: CrdReconciliationMode::default(),
3267            default_owner: None,
3268            profiles: std::collections::HashMap::new(),
3269            schemas: vec![],
3270            roles: vec![],
3271            grants: vec![],
3272            default_privileges: vec![],
3273            memberships: vec![],
3274            retirements: vec![],
3275            approval: None,
3276        };
3277
3278        let names = spec.referenced_secret_names("test-policy");
3279        assert!(names.contains("pg-conn"));
3280        assert_eq!(names.len(), 1);
3281    }
3282
3283    #[test]
3284    fn referenced_secret_names_includes_password_secrets() {
3285        let spec = PostgresPolicySpec {
3286            connection: ConnectionSpec {
3287                secret_ref: Some(SecretReference {
3288                    name: "pg-conn".to_string(),
3289                }),
3290                secret_key: Some("DATABASE_URL".to_string()),
3291                params: None,
3292            },
3293            interval: "5m".to_string(),
3294            suspend: false,
3295            mode: PolicyMode::Apply,
3296            reconciliation_mode: CrdReconciliationMode::default(),
3297            default_owner: None,
3298            profiles: std::collections::HashMap::new(),
3299            schemas: vec![],
3300            roles: vec![
3301                RoleSpec {
3302                    name: "role-a".to_string(),
3303                    external: false,
3304                    login: Some(true),
3305                    password: Some(PasswordSpec {
3306                        secret_ref: Some(SecretReference {
3307                            name: "role-passwords".to_string(),
3308                        }),
3309                        secret_key: Some("role-a".to_string()),
3310                        generate: None,
3311                    }),
3312                    password_valid_until: None,
3313                    superuser: None,
3314                    createdb: None,
3315                    createrole: None,
3316                    inherit: None,
3317                    replication: None,
3318                    bypassrls: None,
3319                    connection_limit: None,
3320                    comment: None,
3321                },
3322                RoleSpec {
3323                    name: "role-b".to_string(),
3324                    external: false,
3325                    login: Some(true),
3326                    password: Some(PasswordSpec {
3327                        secret_ref: Some(SecretReference {
3328                            name: "other-secret".to_string(),
3329                        }),
3330                        secret_key: None,
3331                        generate: None,
3332                    }),
3333                    password_valid_until: None,
3334                    superuser: None,
3335                    createdb: None,
3336                    createrole: None,
3337                    inherit: None,
3338                    replication: None,
3339                    bypassrls: None,
3340                    connection_limit: None,
3341                    comment: None,
3342                },
3343                RoleSpec {
3344                    name: "role-c".to_string(),
3345                    external: false,
3346                    login: None,
3347                    password: None,
3348                    password_valid_until: None,
3349                    superuser: None,
3350                    createdb: None,
3351                    createrole: None,
3352                    inherit: None,
3353                    replication: None,
3354                    bypassrls: None,
3355                    connection_limit: None,
3356                    comment: None,
3357                },
3358            ],
3359            grants: vec![],
3360            default_privileges: vec![],
3361            memberships: vec![],
3362            retirements: vec![],
3363            approval: None,
3364        };
3365
3366        let names = spec.referenced_secret_names("test-policy");
3367        assert!(
3368            names.contains("pg-conn"),
3369            "should include connection secret"
3370        );
3371        assert!(
3372            names.contains("role-passwords"),
3373            "should include role-a password secret"
3374        );
3375        assert!(
3376            names.contains("other-secret"),
3377            "should include role-b password secret"
3378        );
3379        assert_eq!(names.len(), 3);
3380    }
3381
3382    #[test]
3383    fn validate_password_specs_rejects_password_without_login() {
3384        let spec = PostgresPolicySpec {
3385            connection: ConnectionSpec {
3386                secret_ref: Some(SecretReference {
3387                    name: "pg-conn".to_string(),
3388                }),
3389                secret_key: Some("DATABASE_URL".to_string()),
3390                params: None,
3391            },
3392            interval: "5m".to_string(),
3393            suspend: false,
3394            mode: PolicyMode::Apply,
3395            reconciliation_mode: CrdReconciliationMode::default(),
3396            default_owner: None,
3397            profiles: std::collections::HashMap::new(),
3398            schemas: vec![],
3399            roles: vec![RoleSpec {
3400                name: "app-user".to_string(),
3401                external: false,
3402                login: Some(false),
3403                superuser: None,
3404                createdb: None,
3405                createrole: None,
3406                inherit: None,
3407                replication: None,
3408                bypassrls: None,
3409                connection_limit: None,
3410                comment: None,
3411                password: Some(PasswordSpec {
3412                    secret_ref: Some(SecretReference {
3413                        name: "role-passwords".to_string(),
3414                    }),
3415                    secret_key: None,
3416                    generate: None,
3417                }),
3418                password_valid_until: None,
3419            }],
3420            grants: vec![],
3421            default_privileges: vec![],
3422            memberships: vec![],
3423            retirements: vec![],
3424            approval: None,
3425        };
3426
3427        assert!(matches!(
3428            spec.validate_password_specs("test-policy"),
3429            Err(PasswordValidationError::PasswordWithoutLogin { ref role }) if role == "app-user"
3430        ));
3431    }
3432
3433    #[test]
3434    fn validate_password_specs_rejects_password_with_login_omitted() {
3435        let spec = PostgresPolicySpec {
3436            connection: ConnectionSpec {
3437                secret_ref: Some(SecretReference {
3438                    name: "pg-conn".to_string(),
3439                }),
3440                secret_key: Some("DATABASE_URL".to_string()),
3441                params: None,
3442            },
3443            interval: "5m".to_string(),
3444            suspend: false,
3445            mode: PolicyMode::Apply,
3446            reconciliation_mode: CrdReconciliationMode::default(),
3447            default_owner: None,
3448            profiles: std::collections::HashMap::new(),
3449            schemas: vec![],
3450            roles: vec![RoleSpec {
3451                name: "app-user".to_string(),
3452                external: false,
3453                login: None, // omitted, not explicitly false
3454                superuser: None,
3455                createdb: None,
3456                createrole: None,
3457                inherit: None,
3458                replication: None,
3459                bypassrls: None,
3460                connection_limit: None,
3461                comment: None,
3462                password: Some(PasswordSpec {
3463                    secret_ref: Some(SecretReference {
3464                        name: "role-passwords".to_string(),
3465                    }),
3466                    secret_key: None,
3467                    generate: None,
3468                }),
3469                password_valid_until: None,
3470            }],
3471            grants: vec![],
3472            default_privileges: vec![],
3473            memberships: vec![],
3474            retirements: vec![],
3475            approval: None,
3476        };
3477
3478        assert!(matches!(
3479            spec.validate_password_specs("test-policy"),
3480            Err(PasswordValidationError::PasswordWithoutLogin { ref role }) if role == "app-user"
3481        ));
3482    }
3483
3484    #[test]
3485    fn validate_password_specs_rejects_invalid_password_mode() {
3486        let spec = PostgresPolicySpec {
3487            connection: ConnectionSpec {
3488                secret_ref: Some(SecretReference {
3489                    name: "pg-conn".to_string(),
3490                }),
3491                secret_key: Some("DATABASE_URL".to_string()),
3492                params: None,
3493            },
3494            interval: "5m".to_string(),
3495            suspend: false,
3496            mode: PolicyMode::Apply,
3497            reconciliation_mode: CrdReconciliationMode::default(),
3498            default_owner: None,
3499            profiles: std::collections::HashMap::new(),
3500            schemas: vec![],
3501            roles: vec![RoleSpec {
3502                name: "app-user".to_string(),
3503                external: false,
3504                login: Some(true),
3505                superuser: None,
3506                createdb: None,
3507                createrole: None,
3508                inherit: None,
3509                replication: None,
3510                bypassrls: None,
3511                connection_limit: None,
3512                comment: None,
3513                password: Some(PasswordSpec {
3514                    secret_ref: Some(SecretReference {
3515                        name: "role-passwords".to_string(),
3516                    }),
3517                    secret_key: None,
3518                    generate: Some(GeneratePasswordSpec {
3519                        length: Some(32),
3520                        secret_name: None,
3521                        secret_key: None,
3522                    }),
3523                }),
3524                password_valid_until: None,
3525            }],
3526            grants: vec![],
3527            default_privileges: vec![],
3528            memberships: vec![],
3529            retirements: vec![],
3530            approval: None,
3531        };
3532
3533        assert!(matches!(
3534            spec.validate_password_specs("test-policy"),
3535            Err(PasswordValidationError::InvalidPasswordMode { ref role }) if role == "app-user"
3536        ));
3537    }
3538
3539    #[test]
3540    fn validate_password_specs_rejects_invalid_generated_length() {
3541        let spec = PostgresPolicySpec {
3542            connection: ConnectionSpec {
3543                secret_ref: Some(SecretReference {
3544                    name: "pg-conn".to_string(),
3545                }),
3546                secret_key: Some("DATABASE_URL".to_string()),
3547                params: None,
3548            },
3549            interval: "5m".to_string(),
3550            suspend: false,
3551            mode: PolicyMode::Apply,
3552            reconciliation_mode: CrdReconciliationMode::default(),
3553            default_owner: None,
3554            profiles: std::collections::HashMap::new(),
3555            schemas: vec![],
3556            roles: vec![RoleSpec {
3557                name: "app-user".to_string(),
3558                external: false,
3559                login: Some(true),
3560                superuser: None,
3561                createdb: None,
3562                createrole: None,
3563                inherit: None,
3564                replication: None,
3565                bypassrls: None,
3566                connection_limit: None,
3567                comment: None,
3568                password: Some(PasswordSpec {
3569                    secret_ref: None,
3570                    secret_key: None,
3571                    generate: Some(GeneratePasswordSpec {
3572                        length: Some(8),
3573                        secret_name: None,
3574                        secret_key: None,
3575                    }),
3576                }),
3577                password_valid_until: None,
3578            }],
3579            grants: vec![],
3580            default_privileges: vec![],
3581            memberships: vec![],
3582            retirements: vec![],
3583            approval: None,
3584        };
3585
3586        assert!(matches!(
3587            spec.validate_password_specs("test-policy"),
3588            Err(PasswordValidationError::InvalidGeneratedLength { ref role, .. }) if role == "app-user"
3589        ));
3590    }
3591
3592    #[test]
3593    fn validate_password_specs_rejects_invalid_generated_secret_key() {
3594        let spec = PostgresPolicySpec {
3595            connection: ConnectionSpec {
3596                secret_ref: Some(SecretReference {
3597                    name: "pg-conn".to_string(),
3598                }),
3599                secret_key: Some("DATABASE_URL".to_string()),
3600                params: None,
3601            },
3602            interval: "5m".to_string(),
3603            suspend: false,
3604            mode: PolicyMode::Apply,
3605            reconciliation_mode: CrdReconciliationMode::default(),
3606            default_owner: None,
3607            profiles: std::collections::HashMap::new(),
3608            schemas: vec![],
3609            roles: vec![RoleSpec {
3610                name: "app-user".to_string(),
3611                external: false,
3612                login: Some(true),
3613                superuser: None,
3614                createdb: None,
3615                createrole: None,
3616                inherit: None,
3617                replication: None,
3618                bypassrls: None,
3619                connection_limit: None,
3620                comment: None,
3621                password: Some(PasswordSpec {
3622                    secret_ref: None,
3623                    secret_key: None,
3624                    generate: Some(GeneratePasswordSpec {
3625                        length: Some(32),
3626                        secret_name: None,
3627                        secret_key: Some("bad/key".to_string()),
3628                    }),
3629                }),
3630                password_valid_until: None,
3631            }],
3632            grants: vec![],
3633            default_privileges: vec![],
3634            memberships: vec![],
3635            retirements: vec![],
3636            approval: None,
3637        };
3638
3639        assert!(matches!(
3640            spec.validate_password_specs("test-policy"),
3641            Err(PasswordValidationError::InvalidSecretKey { ref role, field, .. })
3642                if role == "app-user" && field == "generate.secretKey"
3643        ));
3644    }
3645
3646    #[test]
3647    fn validate_password_specs_rejects_invalid_generated_secret_name() {
3648        let spec = PostgresPolicySpec {
3649            connection: ConnectionSpec {
3650                secret_ref: Some(SecretReference {
3651                    name: "pg-conn".to_string(),
3652                }),
3653                secret_key: Some("DATABASE_URL".to_string()),
3654                params: None,
3655            },
3656            interval: "5m".to_string(),
3657            suspend: false,
3658            mode: PolicyMode::Apply,
3659            reconciliation_mode: CrdReconciliationMode::default(),
3660            default_owner: None,
3661            profiles: std::collections::HashMap::new(),
3662            schemas: vec![],
3663            roles: vec![RoleSpec {
3664                name: "app-user".to_string(),
3665                external: false,
3666                login: Some(true),
3667                superuser: None,
3668                createdb: None,
3669                createrole: None,
3670                inherit: None,
3671                replication: None,
3672                bypassrls: None,
3673                connection_limit: None,
3674                comment: None,
3675                password: Some(PasswordSpec {
3676                    secret_ref: None,
3677                    secret_key: None,
3678                    generate: Some(GeneratePasswordSpec {
3679                        length: Some(32),
3680                        secret_name: Some("Bad_Name".to_string()),
3681                        secret_key: None,
3682                    }),
3683                }),
3684                password_valid_until: None,
3685            }],
3686            grants: vec![],
3687            default_privileges: vec![],
3688            memberships: vec![],
3689            retirements: vec![],
3690            approval: None,
3691        };
3692
3693        assert!(matches!(
3694            spec.validate_password_specs("test-policy"),
3695            Err(PasswordValidationError::InvalidGeneratedSecretName { ref role, .. }) if role == "app-user"
3696        ));
3697    }
3698
3699    #[test]
3700    fn validate_password_specs_rejects_reserved_generated_secret_key() {
3701        let spec = PostgresPolicySpec {
3702            connection: ConnectionSpec {
3703                secret_ref: Some(SecretReference {
3704                    name: "pg-conn".to_string(),
3705                }),
3706                secret_key: Some("DATABASE_URL".to_string()),
3707                params: None,
3708            },
3709            interval: "5m".to_string(),
3710            suspend: false,
3711            mode: PolicyMode::Apply,
3712            reconciliation_mode: CrdReconciliationMode::default(),
3713            default_owner: None,
3714            profiles: std::collections::HashMap::new(),
3715            schemas: vec![],
3716            roles: vec![RoleSpec {
3717                name: "app-user".to_string(),
3718                external: false,
3719                login: Some(true),
3720                superuser: None,
3721                createdb: None,
3722                createrole: None,
3723                inherit: None,
3724                replication: None,
3725                bypassrls: None,
3726                connection_limit: None,
3727                comment: None,
3728                password: Some(PasswordSpec {
3729                    secret_ref: None,
3730                    secret_key: None,
3731                    generate: Some(GeneratePasswordSpec {
3732                        length: Some(32),
3733                        secret_name: None,
3734                        secret_key: Some("verifier".to_string()),
3735                    }),
3736                }),
3737                password_valid_until: None,
3738            }],
3739            grants: vec![],
3740            default_privileges: vec![],
3741            memberships: vec![],
3742            retirements: vec![],
3743            approval: None,
3744        };
3745
3746        assert!(matches!(
3747            spec.validate_password_specs("test-policy"),
3748            Err(PasswordValidationError::ReservedGeneratedSecretKey { ref role, ref key })
3749                if role == "app-user" && key == "verifier"
3750        ));
3751    }
3752
3753    #[test]
3754    fn plan_crd_generates_valid_schema() {
3755        let crd = PostgresPolicyPlan::crd();
3756        let yaml = serde_yaml::to_string(&crd).expect("CRD should serialize to YAML");
3757        assert!(yaml.contains("pgroles.io"), "group should be pgroles.io");
3758        assert!(yaml.contains("v1alpha1"), "version should be v1alpha1");
3759        assert!(
3760            yaml.contains("PostgresPolicyPlan"),
3761            "kind should be PostgresPolicyPlan"
3762        );
3763        assert!(yaml.contains("pgplan"), "should have shortname pgplan");
3764    }
3765
3766    #[test]
3767    fn plan_phase_display() {
3768        assert_eq!(PlanPhase::Pending.to_string(), "Pending");
3769        assert_eq!(PlanPhase::Approved.to_string(), "Approved");
3770        assert_eq!(PlanPhase::Applying.to_string(), "Applying");
3771        assert_eq!(PlanPhase::Applied.to_string(), "Applied");
3772        assert_eq!(PlanPhase::Failed.to_string(), "Failed");
3773        assert_eq!(PlanPhase::Superseded.to_string(), "Superseded");
3774    }
3775
3776    #[test]
3777    fn plan_phase_default_is_pending() {
3778        assert_eq!(PlanPhase::default(), PlanPhase::Pending);
3779    }
3780
3781    #[test]
3782    fn effective_approval_infers_from_mode() {
3783        let base = PostgresPolicySpec {
3784            connection: ConnectionSpec {
3785                secret_ref: Some(SecretReference {
3786                    name: "test".into(),
3787                }),
3788                secret_key: Some("DATABASE_URL".into()),
3789                params: None,
3790            },
3791            interval: "5m".into(),
3792            suspend: false,
3793            mode: PolicyMode::Apply,
3794            reconciliation_mode: CrdReconciliationMode::Authoritative,
3795            default_owner: None,
3796            profiles: Default::default(),
3797            schemas: vec![],
3798            roles: vec![],
3799            grants: vec![],
3800            default_privileges: vec![],
3801            memberships: vec![],
3802            retirements: vec![],
3803            approval: None,
3804        };
3805
3806        // apply mode with no explicit approval → Auto
3807        assert_eq!(base.effective_approval(), ApprovalMode::Auto);
3808
3809        // plan mode with no explicit approval → Manual
3810        let plan = PostgresPolicySpec {
3811            mode: PolicyMode::Plan,
3812            ..base.clone()
3813        };
3814        assert_eq!(plan.effective_approval(), ApprovalMode::Manual);
3815
3816        // explicit Manual overrides apply mode
3817        let explicit = PostgresPolicySpec {
3818            approval: Some(ApprovalMode::Manual),
3819            ..base.clone()
3820        };
3821        assert_eq!(explicit.effective_approval(), ApprovalMode::Manual);
3822    }
3823
3824    #[test]
3825    fn approval_mode_serde_roundtrip() {
3826        // Deserialize
3827        let manual: ApprovalMode = serde_json::from_str("\"manual\"").unwrap();
3828        assert_eq!(manual, ApprovalMode::Manual);
3829        let auto: ApprovalMode = serde_json::from_str("\"auto\"").unwrap();
3830        assert_eq!(auto, ApprovalMode::Auto);
3831
3832        // Serialize back
3833        let manual_json = serde_json::to_value(&ApprovalMode::Manual).unwrap();
3834        assert_eq!(manual_json, serde_json::Value::String("manual".to_string()));
3835        let auto_json = serde_json::to_value(&ApprovalMode::Auto).unwrap();
3836        assert_eq!(auto_json, serde_json::Value::String("auto".to_string()));
3837    }
3838
3839    #[test]
3840    fn plan_status_default_is_empty() {
3841        let status = PostgresPolicyPlanStatus::default();
3842        assert_eq!(status.phase, PlanPhase::Pending);
3843        assert!(status.conditions.is_empty());
3844        assert!(status.change_summary.is_none());
3845        assert!(status.sql_ref.is_none());
3846        assert!(status.sql_inline.is_none());
3847        assert!(status.computed_at.is_none());
3848        assert!(status.applied_at.is_none());
3849        assert!(status.last_error.is_none());
3850    }
3851
3852    #[test]
3853    fn spec_without_approval_field_deserializes_as_none() {
3854        let json = serde_json::json!({
3855            "connection": {
3856                "secretRef": { "name": "pg-secret" },
3857                "secretKey": "DATABASE_URL"
3858            },
3859            "interval": "5m",
3860            "suspend": false,
3861            "mode": "apply",
3862            "reconciliation_mode": "authoritative"
3863        });
3864
3865        let spec: PostgresPolicySpec =
3866            serde_json::from_value(json).expect("should deserialize without approval field");
3867        assert!(
3868            spec.approval.is_none(),
3869            "approval should be None when omitted"
3870        );
3871        assert_eq!(
3872            spec.effective_approval(),
3873            ApprovalMode::Auto,
3874            "effective_approval should infer Auto from apply mode"
3875        );
3876    }
3877
3878    #[test]
3879    fn status_without_current_plan_ref_deserializes_as_none() {
3880        let json = serde_json::json!({
3881            "conditions": [],
3882            "owned_roles": [],
3883            "owned_schemas": []
3884        });
3885
3886        let status: PostgresPolicyStatus =
3887            serde_json::from_value(json).expect("should deserialize without current_plan_ref");
3888        assert!(
3889            status.current_plan_ref.is_none(),
3890            "current_plan_ref should be None when omitted"
3891        );
3892    }
3893
3894    #[test]
3895    fn effective_approval_explicit_auto_overrides_plan_mode() {
3896        let spec = PostgresPolicySpec {
3897            connection: ConnectionSpec {
3898                secret_ref: Some(SecretReference {
3899                    name: "test".into(),
3900                }),
3901                secret_key: Some("DATABASE_URL".into()),
3902                params: None,
3903            },
3904            interval: "5m".into(),
3905            suspend: false,
3906            mode: PolicyMode::Plan,
3907            reconciliation_mode: CrdReconciliationMode::Authoritative,
3908            default_owner: None,
3909            profiles: Default::default(),
3910            schemas: vec![],
3911            roles: vec![],
3912            grants: vec![],
3913            default_privileges: vec![],
3914            memberships: vec![],
3915            retirements: vec![],
3916            approval: Some(ApprovalMode::Auto),
3917        };
3918
3919        assert_eq!(
3920            spec.effective_approval(),
3921            ApprovalMode::Auto,
3922            "explicit Auto should override Plan mode's default of Manual"
3923        );
3924    }
3925
3926    #[test]
3927    fn plan_phase_rejected_display() {
3928        assert_eq!(PlanPhase::Rejected.to_string(), "Rejected");
3929    }
3930
3931    #[test]
3932    fn plan_phase_all_variants_display() {
3933        let variants = [
3934            PlanPhase::Pending,
3935            PlanPhase::Approved,
3936            PlanPhase::Applying,
3937            PlanPhase::Applied,
3938            PlanPhase::Failed,
3939            PlanPhase::Superseded,
3940            PlanPhase::Rejected,
3941        ];
3942        for variant in &variants {
3943            let display = variant.to_string();
3944            assert!(
3945                !display.is_empty(),
3946                "PlanPhase::{variant:?} should have non-empty Display output"
3947            );
3948        }
3949    }
3950
3951    #[test]
3952    fn plan_status_defaults() {
3953        let status = PostgresPolicyPlanStatus::default();
3954        assert_eq!(status.phase, PlanPhase::Pending);
3955        assert!(status.conditions.is_empty());
3956        assert!(status.sql_ref.is_none());
3957        assert!(status.sql_hash.is_none());
3958        assert!(status.sql_inline.is_none());
3959        assert!(!status.sql_truncated);
3960        assert!(status.redacted_sql_hash.is_none());
3961        assert!(status.sql_original_bytes.is_none());
3962        assert!(status.sql_stored_bytes.is_none());
3963        assert!(status.change_summary.is_none());
3964        assert!(status.computed_at.is_none());
3965        assert!(status.applied_at.is_none());
3966        assert!(status.last_error.is_none());
3967    }
3968
3969    #[test]
3970    fn sql_ref_missing_compression_deserializes_as_uncompressed_legacy_shape() {
3971        let json = serde_json::json!({
3972            "name": "legacy-plan-sql",
3973            "key": "plan.sql"
3974        });
3975
3976        let sql_ref: SqlRef = serde_json::from_value(json).expect("legacy SqlRef should decode");
3977
3978        assert_eq!(sql_ref.name, "legacy-plan-sql");
3979        assert_eq!(sql_ref.key, "plan.sql");
3980        assert_eq!(sql_ref.compression, None);
3981    }
3982
3983    #[test]
3984    fn plan_spec_camel_case_serialization() {
3985        let spec = PostgresPolicyPlanSpec {
3986            policy_ref: PolicyPlanRef {
3987                name: "my-policy".into(),
3988            },
3989            policy_generation: 3,
3990            reconciliation_mode: CrdReconciliationMode::Authoritative,
3991            owned_roles: vec!["role-a".into()],
3992            owned_schemas: vec!["public".into()],
3993            managed_database_identity: "ns/secret/key".into(),
3994        };
3995
3996        let json = serde_json::to_value(&spec).expect("should serialize to JSON");
3997        let obj = json.as_object().expect("should be a JSON object");
3998
3999        assert!(
4000            obj.contains_key("policyRef"),
4001            "should use camelCase: policyRef"
4002        );
4003        assert!(
4004            obj.contains_key("policyGeneration"),
4005            "should use camelCase: policyGeneration"
4006        );
4007        assert!(
4008            obj.contains_key("reconciliationMode"),
4009            "should use camelCase: reconciliationMode"
4010        );
4011        assert!(
4012            obj.contains_key("ownedRoles"),
4013            "should use camelCase: ownedRoles"
4014        );
4015        assert!(
4016            obj.contains_key("ownedSchemas"),
4017            "should use camelCase: ownedSchemas"
4018        );
4019        assert!(
4020            obj.contains_key("managedDatabaseIdentity"),
4021            "should use camelCase: managedDatabaseIdentity"
4022        );
4023    }
4024}
pgroles_operator/crd.rs

pgroles_operator/
crd.rs
