feat(python): SQLAlchemy layer, order and match queries, and row-level encrypt and decrypt

languages/python/.gitignore

@@ -4,3 +4,7 @@
 *.so
 *.pyd
 .pytest_cache/
+# The CS_* credentials for `mise run test:live`, if `stash env` writes them
+# here. The file contains a client key and an access key, so it must never
+# be committed.
+.stackencrypt-live.env

languages/python/crates/core/fuzz/fuzz_targets/value_frame.rs

@@ -71,6 +71,7 @@ enum Host {
     Date(i64),
     Timestamp(i128),
     Decimal { mantissa: i128, scale: u32 },
+    Null,
 }
 
 impl Host {
@@ -87,6 +88,7 @@ impl Host {
                 mantissa: *mantissa,
                 scale: *scale,
             },
+            Self::Null => Value::Null,
         }
     }
 }
@@ -107,6 +109,7 @@ fn expected(kind: Option<Kind>, value: Value<'_>) -> Scalar {
             mantissa,
             scale: u8::try_from(scale).expect("an accepted scale"),
         },
+        (_, Value::Null) => Scalar::Null,
     }
 }
 

languages/python/crates/core/src/decrypt.rs

@@ -459,13 +459,15 @@ pub(crate) mod tests {
         Field {
             name: name.to_owned(),
             eql: Some((eql.to_owned(), kind)),
+            nullable: false,
         }
     }
 
     fn passthrough(name: &str) -> Field {
         Field {
             name: name.to_owned(),
             eql: None,
+            nullable: false,
         }
     }
 
@@ -533,6 +535,7 @@ pub(crate) mod tests {
                         row.passthrough(field, Value::Int(*i)).unwrap()
                     }
                     Column::Passthrough(other) => panic!("not an age: {other:?}"),
+                    Column::Null => panic!("no field here is nullable"),
                 }
             }
             rows.push(row);
@@ -717,4 +720,117 @@ pub(crate) mod tests {
         assert_eq!(refused.code, "decrypt_kind_mismatch", "{refused:?}");
         assert_eq!(refused.field.as_deref(), Some("email"));
     }
+
+    /// `people` with a nullable sealed field and a nullable passthrough.
+    fn people() -> ModelPlan {
+        ModelPlan::new(
+            "people",
+            vec![
+                Field {
+                    name: "city".to_owned(),
+                    eql: Some(("TextEq".to_owned(), Kind::String)),
+                    nullable: true,
+                },
+                Field {
+                    name: "email".to_owned(),
+                    eql: Some(("TextEq".to_owned(), Kind::String)),
+                    nullable: false,
+                },
+                Field {
+                    name: "note".to_owned(),
+                    eql: None,
+                    nullable: true,
+                },
+            ],
+        )
+        .unwrap()
+    }
+
+    fn people_rows<'a>(rows: &[(Value<'a>, &'a str, Value<'a>)]) -> ValueFrame {
+        ValueFrame::of(Source(
+            rows.iter()
+                .map(|(city, email, note)| {
+                    vec![
+                        ("city", Some(Kind::String), *city),
+                        ("email", Some(Kind::String), Value::Text(email)),
+                        ("note", None, *note),
+                    ]
+                })
+                .collect(),
+        ))
+        .unwrap()
+    }
+
+    #[test]
+    fn a_null_in_a_nullable_field_is_stored_as_sql_null_and_opens_as_null() {
+        let cipher = cipher();
+        let plan = people();
+        let source = people_rows(&[
+            (Value::Null, "a@example.com", Value::Null),
+            (Value::Text("Leeds"), "b@example.com", Value::Text("hi")),
+        ]);
+
+        let stored = run(encrypt_with(&cipher.default_keyset(), &plan, source, 2)).unwrap();
+
+        assert!(matches!(stored[0][0], Column::Null));
+        assert!(matches!(
+            stored[0][2],
+            Column::Null | Column::Passthrough(Scalar::Null)
+        ));
+        assert!(matches!(stored[1][0], Column::Eql(_)));
+
+        let mut rows = Rows::default();
+        for columns in &stored {
+            let mut row = Row::new();
+            for (field, column) in plan.fields().iter().zip(columns) {
+                match column {
+                    Column::Eql(json) => row.eql(&plan, field, json.as_bytes()).unwrap(),
+                    Column::Passthrough(Scalar::Text(text)) => {
+                        row.passthrough(field, Value::Text(text)).unwrap()
+                    }
+                    // A passthrough is always in the row, as a null here.
+                    Column::Passthrough(Scalar::Null) => {
+                        row.passthrough(field, Value::Null).unwrap()
+                    }
+                    // A nullable EQL field's SQL NULL: no entry.
+                    Column::Null if field.eql.is_some() => {}
+                    Column::Null => row.passthrough(field, Value::Null).unwrap(),
+                    Column::Passthrough(other) => panic!("not a note: {other:?}"),
+                }
+            }
+            rows.push(row);
+        }
+        let opened = run(decrypt_with(&cipher, &plan, rows)).unwrap();
+
+        assert_eq!(
+            opened,
+            vec![
+                vec![
+                    Scalar::Null,
+                    Scalar::Text("a@example.com".to_owned()),
+                    Scalar::Null
+                ],
+                vec![
+                    Scalar::Text("Leeds".to_owned()),
+                    Scalar::Text("b@example.com".to_owned()),
+                    Scalar::Text("hi".to_owned()),
+                ],
+            ]
+        );
+    }
+
+    #[test]
+    fn a_null_in_a_field_that_is_not_nullable_is_refused() {
+        let cipher = cipher();
+        let source = ValueFrame::of(Source(vec![vec![
+            ("city", Some(Kind::String), Value::Text("Leeds")),
+            ("email", Some(Kind::String), Value::Null),
+            ("note", None, Value::Text("hi")),
+        ]]))
+        .unwrap();
+
+        let refused = run(encrypt_with(&cipher.default_keyset(), &people(), source, 1));
+
+        assert!(refused.is_err(), "{refused:?}");
+    }
 }

languages/python/crates/core/src/encrypt.rs

@@ -30,6 +30,9 @@ pub struct Field {
     pub name: String,
     /// The EQL type and its kind, or `None` for a passthrough.
     pub eql: Option<(String, Kind)>,
+    /// Whether the field can be missing: Python's `Optional`. A missing
+    /// value is written as a null, and comes back as [`Column::Null`].
+    pub nullable: bool,
 }
 
 /// A model's plan, built once per `Cipher` and read on every call.
@@ -53,7 +56,11 @@ impl ModelPlan {
                     .eql
                     .as_ref()
                     .map_or(plan::Column::Passthrough, |(name, _)| {
-                        plan::Column::Eql(name)
+                        if field.nullable {
+                            plan::Column::NullableEql(name)
+                        } else {
+                            plan::Column::Eql(name)
+                        }
                     });
                 (field.name.as_str(), column)
             })
@@ -90,6 +97,8 @@ pub enum Column {
     Eql(String),
     /// The echoed value of a passthrough field.
     Passthrough(Scalar),
+    /// A nullable field's missing value: SQL `NULL`, Python's `None`.
+    Null,
 }
 
 /// Encrypt `rows` models, written as one frame, under `plan`. Returns one
@@ -152,10 +161,14 @@ fn read_row(plan: &ModelPlan, row: StackCipherText) -> Result<Vec<Column>, Refus
     };
     let mut columns = Vec::with_capacity(plan.fields.len());
     for field in &plan.fields {
-        let (_, outputs) = entries
-            .iter()
-            .find(|(name, _)| name == &field.name)
-            .ok_or_else(|| engine_fault("a row lacks a declared field"))?;
+        let Some((_, outputs)) = entries.iter().find(|(name, _)| name == &field.name) else {
+            // A nullable EQL field's null produces no entry.
+            if field.nullable {
+                columns.push(Column::Null);
+                continue;
+            }
+            return Err(engine_fault("a row lacks a declared field"));
+        };
         let CipherText::Map(outputs) = outputs else {
             return Err(engine_fault("a field is not a map of outputs"));
         };

languages/python/crates/core/src/plan.rs

@@ -18,6 +18,10 @@ use crate::targets;
 pub enum Column<'a> {
     /// A field sealed as the EQL type of this name, such as `TextEq`.
     Eql(&'a str),
+    /// As [`Column::Eql`], for a field that can be missing. The engine
+    /// stores nothing for a null (`"null": "sql"`), so the column holds SQL
+    /// `NULL`, and opens a missing value as a null.
+    NullableEql(&'a str),
     /// A field stored as it is.
     Passthrough,
 }
@@ -90,10 +94,20 @@ struct Spec<'a> {
 
 impl Serialize for Spec<'_> {
     fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        let mut spec = serializer.serialize_struct("Spec", 2)?;
+        // The frame needs each container's exact length.
+        let len = if matches!(self.column, Column::NullableEql(_)) {
+            3
+        } else {
+            2
+        };
+        let mut spec = serializer.serialize_struct("Spec", len)?;
         spec.serialize_field("context", &self.context)?;
         match self.column {
             Column::Eql(name) => spec.serialize_field("target", name)?,
+            Column::NullableEql(name) => {
+                spec.serialize_field("target", name)?;
+                spec.serialize_field("null", "sql")?;
+            }
             Column::Passthrough => spec.serialize_field("outputs", &["passthrough"])?,
         }
         spec.end()