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

pgroles_operator/
crd.rs
